nautilus-network 0.66.0

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

//! HTTP wire behavior and connection lifecycle regressions.

use std::{
    collections::HashMap,
    io::{self, ErrorKind},
    sync::{
        Arc,
        atomic::{AtomicUsize, Ordering},
    },
    time::Duration,
};

use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
use bytes::Bytes;
use http::Method;
use rstest::rstest;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

use super::{HttpClient, HttpClientError, HttpRedirectPolicy, HttpResponse};

#[rstest]
#[case(301)]
#[case(302)]
#[case(303)]
#[case(307)]
#[case(308)]
#[tokio::test]
async fn rejected_redirect_preserves_response_without_contacting_destination(#[case] status: u16) {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let destination = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let target = destination.local_addr().unwrap();
    let destination_requests = Arc::new(AtomicUsize::new(0));
    let requests = Arc::clone(&destination_requests);

    let destination_task = tokio::spawn(async move {
        let (mut stream, _) = destination.accept().await.unwrap();
        requests.fetch_add(1, Ordering::SeqCst);
        read_headers(&mut stream).await;
        stream
            .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")
            .await
            .unwrap();
    });

    let origin_task = tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.unwrap();
        let headers = read_headers(&mut stream).await;
        let mut body = [0; 7];
        stream.read_exact(&mut body).await.unwrap();
        stream
            .write_all(format!("HTTP/1.1 {status} Redirect\r\nContent-Length: 4\r\nLocation: http://{target}/next\r\n\r\nstop").as_bytes())
            .await
            .unwrap();
        (headers, body)
    });
    let client = HttpClient::builder()
        .redirect_policy(HttpRedirectPolicy::Reject)
        .headers(HashMap::from([("x-api-key".into(), "test-key".into())]))
        .header_keys(vec!["location".into()])
        .use_system_proxy(false)
        .timeout_secs(3)
        .build()
        .unwrap();
    let response = send(
        &client,
        Method::POST,
        format!("http://{addr}/start"),
        Some(b"payload".to_vec()),
    )
    .await
    .unwrap();
    let (headers, body) = origin_task.await.unwrap();
    destination_task.abort();

    assert_eq!(response.status.as_u16(), status);
    assert_eq!(response.body.as_ref(), b"stop");
    assert_eq!(
        response.headers,
        HashMap::from([("location".into(), format!("http://{target}/next"))])
    );
    assert!(headers.starts_with("POST /start HTTP/1.1\r\n"));
    assert!(headers.contains("\r\nx-api-key: test-key\r\n"));
    assert_eq!(&body, b"payload");
    assert_eq!(destination_requests.load(Ordering::SeqCst), 0);
}

#[tokio::test]
async fn pooled_requests_consume_complete_bodies() {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let peer = tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.unwrap();
        for _ in 0..3 {
            let request = read_headers(&mut stream).await;
            assert!(request.starts_with("GET /pool?asset=BTC HTTP/1.1\r\n"));
            stream.write_all(b"HTTP/1.1 206 Partial Content\r\nTransfer-Encoding: chunked\r\nX-Receipt: receipt-53\r\n\r\n3\r\nabc\r\n4\r\ndefg\r\n0\r\n\r\n").await.unwrap();
        }
    });
    let client = HttpClient::builder()
        .header_keys(vec!["x-receipt".into()])
        .use_system_proxy(false)
        .timeout_secs(3)
        .build()
        .unwrap();

    for _ in 0..3 {
        let response = send(
            &client,
            Method::GET,
            format!("http://{addr}/pool?asset=BTC"),
            None,
        )
        .await
        .unwrap();
        assert_eq!(response.status.as_u16(), 206);
        assert_eq!(
            response.headers,
            HashMap::from([("x-receipt".into(), "receipt-53".into())])
        );
        assert_eq!(response.body, Bytes::from_static(b"abcdefg"));
    }
    peer.await.unwrap();
}

#[rstest]
#[case::moved(301, "POST", "GET", false)]
#[case::found(302, "POST", "GET", false)]
#[case::see_other(303, "PUT", "GET", false)]
#[case::temporary(307, "POST", "POST", true)]
#[case::permanent(308, "PUT", "PUT", true)]
#[case::put_found(302, "PUT", "PUT", true)]
#[tokio::test]
async fn redirect_method_body_and_credentials_match(
    #[case] status: u16,
    #[case] initial: &str,
    #[case] redirected: &str,
    #[case] retained: bool,
) {
    {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let destination = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let target = destination.local_addr().unwrap();
        let peer = tokio::spawn(async move {
            let (mut stream, _) = listener.accept().await.unwrap();
            read_headers(&mut stream).await;
            let mut body = [0; 7];
            stream.read_exact(&mut body).await.unwrap();
            assert_eq!(&body, b"payload");
            stream.write_all(format!("HTTP/1.1 {status} Redirect\r\nContent-Length: 0\r\nLocation: http://{target}/next\r\n\r\n").as_bytes()).await.unwrap();
            let (mut stream, _) = destination.accept().await.unwrap();
            let headers = read_headers(&mut stream).await;
            let mut body = Vec::new();
            if retained {
                body.resize(7, 0);
                stream.read_exact(&mut body).await.unwrap();
            }
            stream
                .write_all(b"HTTP/1.1 201 Created\r\nContent-Length: 2\r\n\r\nok")
                .await
                .unwrap();
            (headers, body)
        });
        let defaults = HashMap::from([
            ("authorization".into(), "Bearer private-27".into()),
            ("cookie".into(), "session=private-91".into()),
            ("content-type".into(), "application/octet-stream".into()),
        ]);
        let client = HttpClient::builder()
            .headers(defaults.clone())
            .use_system_proxy(false)
            .timeout_secs(3)
            .build()
            .unwrap();
        let response = send(
            &client,
            initial.parse().unwrap(),
            format!("http://{addr}/start#fragment"),
            Some(b"payload".to_vec()),
        )
        .await
        .unwrap();
        let (headers, body) = peer.await.unwrap();
        assert_eq!(response.status.as_u16(), 201);
        assert_eq!(response.body, Bytes::from_static(b"ok"));
        assert_eq!(response.headers, HashMap::new());
        assert!(
            headers.starts_with(&format!("{redirected} /next HTTP/1.1\r\n")),
            "{headers}"
        );
        assert!(!headers.contains("private-"), "{headers}");
        assert!(
            headers.contains(&format!("referer: http://{addr}/start\r\n")),
            "{headers}"
        );
        assert_eq!(
            headers.contains("content-type: application/octet-stream"),
            retained
        );
        assert_eq!(
            body,
            if retained {
                b"payload".to_vec()
            } else {
                vec![]
            }
        );
    }
}

#[rstest]
#[case::ten_redirects(10, true)]
#[case::eleven_redirects(11, false)]
#[tokio::test]
async fn redirect_limit_preserves_same_origin_credentials(
    #[case] redirects: usize,
    #[case] success: bool,
) {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let peer = tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.unwrap();
        for step in 0..=redirects.min(10) {
            let headers = read_headers(&mut stream).await;
            assert!(
                headers.starts_with(&format!("GET /{step} HTTP/1.1\r\n")),
                "{headers}"
            );
            assert!(
                headers.contains("authorization: Bearer token-37\r\n"),
                "{headers}"
            );
            assert!(headers.contains("cookie: session=91\r\n"), "{headers}");
            let response = if step == redirects {
                "HTTP/1.1 200 OK\r\nContent-Length: 4\r\n\r\ndone".to_string()
            } else {
                format!(
                    "HTTP/1.1 302 Found\r\nContent-Length: 0\r\nLocation: /{}\r\n\r\n",
                    step + 1
                )
            };
            stream.write_all(response.as_bytes()).await.unwrap();
        }
    });
    let client = HttpClient::builder()
        .headers(HashMap::from([
            ("authorization".into(), "Bearer token-37".into()),
            ("cookie".into(), "session=91".into()),
        ]))
        .use_system_proxy(false)
        .timeout_secs(3)
        .build()
        .unwrap();

    let result = send(&client, Method::GET, format!("http://{addr}/0"), None).await;
    tokio::time::timeout(Duration::from_secs(3), peer)
        .await
        .unwrap()
        .unwrap();

    if success {
        let response = result.unwrap();
        assert_eq!(response.status.as_u16(), 200);
        assert_eq!(response.headers, HashMap::new());
        assert_eq!(response.body.as_ref(), b"done");
    } else {
        assert!(
            matches!(result, Err(HttpClientError::Error(ref message)) if message == "too many redirects"),
            "{result:?}"
        );
    }
}

#[tokio::test]
async fn redirect_rejects_unsupported_scheme() {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let peer = tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.unwrap();
        read_headers(&mut stream).await;
        stream.write_all(b"HTTP/1.1 302 Found\r\nContent-Length: 0\r\nLocation: ftp://127.0.0.1/file\r\n\r\n").await.unwrap();
    });
    let client = HttpClient::builder()
        .use_system_proxy(false)
        .timeout_secs(3)
        .build()
        .unwrap();

    let result = send(&client, Method::GET, format!("http://{addr}/start"), None).await;
    peer.await.unwrap();

    assert!(
        matches!(result, Err(HttpClientError::Error(ref message)) if message == "unsupported redirect URL scheme"),
        "{result:?}"
    );
}

#[rstest]
#[case::buffered(false)]
#[case::streamed(true)]
#[tokio::test]
async fn truncated_body_returns_transport_error(
    #[case] streamed: bool,
    #[values(false, true)] redact_url: bool,
) {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let peer = tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.unwrap();
        read_headers(&mut stream).await;
        stream
            .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 8\r\n\r\nshort")
            .await
            .unwrap();
        stream.shutdown().await.unwrap();
    });
    let client = HttpClient::builder()
        .redact_url(redact_url)
        .use_system_proxy(false)
        .timeout_secs(3)
        .build()
        .unwrap();
    let url = format!("http://{addr}/truncated");
    let error = if streamed {
        let mut response = client.get_stream(url).await.unwrap();
        loop {
            match response.chunk().await {
                Ok(Some(_)) => {}
                Ok(None) => panic!("truncated body must not end successfully"),
                Err(e) => break e,
            }
        }
    } else {
        send(&client, Method::GET, url, None).await.unwrap_err()
    };
    peer.await.unwrap();

    let HttpClientError::TransportError(message) = error else {
        panic!("expected transport error, was {error:?}");
    };
    assert!(
        message.contains("end of file before message length reached"),
        "{message}"
    );
    assert_eq!(
        message.contains(&format!(" for origin (http://{addr})")),
        !redact_url,
        "{message}"
    );
    assert_eq!(
        message.contains(&addr.to_string()),
        !redact_url,
        "{message}"
    );
    assert!(!message.contains("/truncated"), "{message}");
}

#[rstest]
#[tokio::test]
async fn streamed_response_url_policy_redacts_debug_and_body_errors(
    #[values(false, true)] redact_url: bool,
) {
    const USERINFO_SECRET: &str = "stream-userinfo-secret";
    const PATH_SECRET: &str = "stream-path-secret";
    const FRAGMENT_SECRET: &str = "stream-fragment-secret";
    const QUERY_SECRET: &str = "stream-query-secret";
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();

    let peer = tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.unwrap();
        let request = read_headers(&mut stream).await;
        stream
            .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 8\r\n\r\nshort")
            .await
            .unwrap();
        stream.shutdown().await.unwrap();
        request
    });

    let client = HttpClient::builder()
        .redact_url(redact_url)
        .use_system_proxy(false)
        .timeout_secs(3)
        .build()
        .unwrap();

    let mut response = client
        .get_stream(format!("http://stream-user:{USERINFO_SECRET}@{addr}/{PATH_SECRET}?api_key={QUERY_SECRET}#{FRAGMENT_SECRET}"))
        .await
        .unwrap();

    let debug = format!("{response:?}");

    let error = loop {
        match response.chunk().await {
            Ok(Some(_)) => {}
            Ok(None) => panic!("truncated body must not end successfully"),
            Err(e) => break e,
        }
    };

    let request = peer.await.unwrap();

    assert!(request.starts_with(&format!(
        "GET /{PATH_SECRET}?api_key={QUERY_SECRET} HTTP/1.1\r\n"
    )));

    let authorization = request
        .lines()
        .find_map(|line| {
            let (name, value) = line.split_once(':')?;
            name.eq_ignore_ascii_case("authorization")
                .then_some(value.trim())
        })
        .unwrap();

    let expected_authorization = format!(
        "Basic {}",
        BASE64.encode(format!("stream-user:{USERINFO_SECRET}"))
    );
    assert_eq!(authorization, expected_authorization);
    assert!(
        matches!(error, HttpClientError::TransportError(_)),
        "{error:?}"
    );

    for rendered in [debug, error.to_string(), format!("{error:?}")] {
        assert_eq!(
            rendered.contains(&format!("http://{addr}")),
            !redact_url,
            "{rendered}"
        );

        for secret in [USERINFO_SECRET, PATH_SECRET, QUERY_SECRET, FRAGMENT_SECRET] {
            assert!(!rendered.contains(secret), "{rendered}");
        }
    }
}

#[tokio::test]
async fn body_deadline_overrides_default_and_closes_connection() {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let peer = tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.unwrap();
        read_headers(&mut stream).await;
        stream
            .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nx")
            .await
            .unwrap();
        let mut byte = [0];
        assert_eq!(
            tokio::time::timeout(Duration::from_secs(4), stream.read(&mut byte))
                .await
                .unwrap()
                .unwrap(),
            0
        );
    });
    let client = HttpClient::builder()
        .timeout_secs(20)
        .use_system_proxy(false)
        .build()
        .unwrap();
    let result = client
        .get(format!("http://{addr}/slow"), None, None, Some(1), None)
        .await;
    assert!(
        matches!(result, Err(HttpClientError::TimeoutError(_))),
        "{result:?}"
    );
    peer.await.unwrap();
}

#[tokio::test]
async fn http_proxy_preserves_absolute_target_and_authentication() {
    const USERNAME: &str = "user";
    const PASSWORD: &str = "secret";

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let peer = tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.unwrap();
        let headers = read_headers(&mut stream).await;
        stream
            .write_all(b"HTTP/1.1 202 Accepted\r\nContent-Length: 7\r\n\r\nproxied")
            .await
            .unwrap();
        headers
    });
    let proxy = format!("http://{USERNAME}:{PASSWORD}@{addr}");
    let client = HttpClient::builder()
        .proxy_url(proxy.clone())
        .timeout_secs(3)
        .build()
        .unwrap();
    let response = send(
        &client,
        Method::GET,
        "http://127.0.0.1:9/target?asset=BTC".into(),
        None,
    )
    .await
    .unwrap();
    let headers = peer.await.unwrap();
    assert!(
        headers.starts_with("GET http://127.0.0.1:9/target?asset=BTC HTTP/1.1\r\n"),
        "{headers}"
    );
    assert!(headers.contains("host: 127.0.0.1:9\r\n"), "{headers}");
    let expected_auth = format!(
        "proxy-authorization: Basic {}\r\n",
        BASE64.encode(format!("{USERNAME}:{PASSWORD}"))
    );
    assert!(headers.contains(&expected_auth), "{headers}");
    assert_eq!(response.status.as_u16(), 202);
    assert_eq!(response.headers, HashMap::new());
    assert_eq!(response.body, Bytes::from_static(b"proxied"));
}

#[tokio::test]
async fn canceled_request_releases_partial_response() {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let (ready, received) = tokio::sync::oneshot::channel();

    let peer = tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.unwrap();
        read_headers(&mut stream).await;
        stream
            .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nx")
            .await
            .unwrap();
        ready.send(()).unwrap();
        let mut byte = [0];
        let read = tokio::time::timeout(Duration::from_secs(3), stream.read(&mut byte))
            .await
            .unwrap();

        // Closing a socket with unread received bytes sends a reset instead of EOF on some
        // platforms; both mean the canceled request released the connection
        assert!(
            matches!(
                read.as_ref().map_err(io::Error::kind),
                Ok(0) | Err(ErrorKind::ConnectionReset)
            ),
            "connection was not released: {read:?}"
        );
    });
    let client = HttpClient::builder()
        .use_system_proxy(false)
        .build()
        .unwrap();

    let request = tokio::spawn(async move {
        client
            .get(format!("http://{addr}/body"), None, None, None, None)
            .await
    });
    received.await.unwrap();
    request.abort();
    assert!(request.await.unwrap_err().is_cancelled());
    peer.await.unwrap();
}

#[tokio::test]
async fn streamed_response_exceeds_buffered_limit() {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let peer = tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.unwrap();
        read_headers(&mut stream).await;
        stream
            .write_all(b"HTTP/1.1 206 Partial Content\r\nContent-Length: 8\r\n\r\nabcdefgh")
            .await
            .unwrap();
    });
    let mut client = HttpClient::builder()
        .use_system_proxy(false)
        .timeout_secs(3)
        .build()
        .unwrap();
    client.client.max_response_bytes = 4;
    let mut response = client
        .get_stream(format!("http://{addr}/large"))
        .await
        .unwrap();
    let status = response.status();
    let mut body = Vec::new();
    while let Some(chunk) = response.chunk().await.unwrap() {
        body.extend_from_slice(&chunk);
    }
    peer.await.unwrap();
    assert_eq!(status, http::StatusCode::PARTIAL_CONTENT);
    assert_eq!(body, b"abcdefgh");
}

#[rstest]
#[case::drop(false)]
#[case::timeout(true)]
#[tokio::test]
async fn streamed_response_releases_partial_body(#[case] timeout: bool) {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let peer = tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.unwrap();
        read_headers(&mut stream).await;
        stream
            .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nx")
            .await
            .unwrap();
        let mut byte = [0];
        assert_eq!(
            tokio::time::timeout(Duration::from_secs(4), stream.read(&mut byte))
                .await
                .unwrap()
                .unwrap(),
            0
        );
    });
    let client = HttpClient::builder()
        .use_system_proxy(false)
        .timeout_secs(1)
        .build()
        .unwrap();
    let mut response = client
        .get_stream(format!("http://{addr}/slow"))
        .await
        .unwrap();
    assert_eq!(
        response.chunk().await.unwrap(),
        Some(Bytes::from_static(b"x"))
    );

    if timeout {
        tokio::time::sleep(Duration::from_millis(1050)).await;
        assert!(matches!(
            response.chunk().await,
            Err(HttpClientError::TimeoutError(_))
        ));
    }
    drop(response);
    peer.await.unwrap();
}

#[cfg(all(unix, not(target_os = "android"), not(target_vendor = "apple")))]
#[rstest]
fn platform_tls_http2_and_protocol_retries() {
    use http::StatusCode;

    let Some(config) = tls_server_config() else {
        run_tls_child("http::tests::platform_tls_http2_and_protocol_retries", &[]);
        return;
    };

    tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap()
        .block_on(async {
            {
                for (reason, refused, expected_requests, success) in [
                    (h2::Reason::REFUSED_STREAM, 2, 3, true),
                    (h2::Reason::REFUSED_STREAM, 3, 3, false),
                    (h2::Reason::INTERNAL_ERROR, 1, 1, false),
                ] {
                    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
                    let addr = listener.local_addr().unwrap();
                    let attempts = Arc::new(AtomicUsize::new(0));
                    let count = Arc::clone(&attempts);
                    let acceptor = tokio_rustls::TlsAcceptor::from(Arc::clone(&config));

                    let peer = tokio::spawn(async move {
                        let (stream, _) = listener.accept().await.unwrap();
                        let stream = acceptor.accept(stream).await.unwrap();
                        assert_eq!(stream.get_ref().1.alpn_protocol(), Some(b"h2".as_slice()));
                        let mut connection = h2::server::handshake(stream).await.unwrap();
                        while let Some(request) = connection.accept().await {
                            let (request, mut respond) = request.unwrap();
                            assert_eq!(request.method(), Method::POST);
                            assert_eq!(request.uri().path(), "/order");

                            if count.fetch_add(1, Ordering::SeqCst) < refused {
                                respond.send_reset(reason);
                            } else {
                                let response = http::Response::builder()
                                    .status(StatusCode::CREATED)
                                    .body(())
                                    .unwrap();
                                respond
                                    .send_response(response, false)
                                    .unwrap()
                                    .send_data(Bytes::from_static(b"accepted-73"), true)
                                    .unwrap();
                            }
                        }
                    });
                    let client = HttpClient::builder()
                        .use_system_proxy(false)
                        .timeout_secs(3)
                        .build()
                        .unwrap();
                    let result = send(
                        &client,
                        Method::POST,
                        format!("https://localhost:{}/order", addr.port()),
                        Some(b"order-83".to_vec()),
                    )
                    .await;
                    assert_eq!(
                        attempts.load(Ordering::SeqCst),
                        expected_requests,
                        "{result:?}"
                    );

                    if success {
                        let response = result.unwrap();
                        assert_eq!(response.status.as_u16(), 201);
                        assert_eq!(response.headers, HashMap::new());
                        assert_eq!(response.body, Bytes::from_static(b"accepted-73"));
                    } else {
                        assert!(
                            matches!(result, Err(HttpClientError::TransportError(_))),
                            "{result:?}"
                        );
                    }
                    peer.abort();
                    assert!(peer.await.unwrap_err().is_cancelled());
                }
            }
        });
}

#[cfg(all(unix, not(target_os = "android"), not(target_vendor = "apple")))]
#[rstest]
fn proxy_authentication_survives_tls_redirects_without_reaching_origin() {
    let Some(config) = tls_server_config() else {
        run_tls_child(
            "http::tests::proxy_authentication_survives_tls_redirects_without_reaching_origin",
            &[("RUST_TEST_THREADS", Some("1"))],
        );
        return;
    };

    tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap()
        .block_on(async {
            tokio::time::timeout(Duration::from_secs(5), async {
                let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
                let addr = listener.local_addr().unwrap();
                let mut config = (*config).clone();
                config.alpn_protocols.clear();
                let acceptor = tokio_rustls::TlsAcceptor::from(Arc::new(config));

                let peer = tokio::spawn(async move {
                    let (mut stream, _) = listener.accept().await.unwrap();
                    let initial = read_headers(&mut stream).await;
                    stream
                        .write_all(b"HTTP/1.1 302 Found\r\nLocation: https://localhost:443/secure\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
                        .await
                        .unwrap();
                    drop(stream);

                    let (mut stream, _) = listener.accept().await.unwrap();
                    let connect = read_headers(&mut stream).await;
                    stream
                        .write_all(b"HTTP/1.1 200 Connection Established\r\n\r\n")
                        .await
                        .unwrap();
                    let mut stream = acceptor.accept(stream).await.unwrap();
                    let origin = read_headers(&mut stream).await;
                    stream
                        .write_all(b"HTTP/1.1 302 Found\r\nLocation: http://localhost:2/final\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
                        .await
                        .unwrap();
                    drop(stream);

                    let (mut stream, _) = listener.accept().await.unwrap();
                    let final_hop = read_headers(&mut stream).await;
                    stream
                        .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 8\r\nConnection: close\r\n\r\naccepted")
                        .await
                        .unwrap();
                    [initial, connect, origin, final_hop]
                });

                let client = HttpClient::builder()
                    .use_system_proxy(false)
                    .proxy_url(format!("http://user:secret@{addr}"))
                    .timeout_secs(3)
                    .build()
                    .unwrap();
                let headers = HashMap::from([
                    ("authorization".into(), "Bearer origin-19".into()),
                    ("cookie".into(), "session=origin-23".into()),
                ]);
                let response = client
                    .request(
                        Method::GET,
                        "http://localhost:1/start".into(),
                        None,
                        Some(headers),
                        None,
                        None,
                        None,
                    )
                    .await
                    .unwrap();
                let requests = peer.await.unwrap();
                let expected_auth = format!("Basic {}", BASE64.encode("user:secret"));

                for (index, (request, target)) in requests
                    .iter()
                    .zip([
                        "GET http://localhost:1/start HTTP/1.1\r\n",
                        "CONNECT localhost:443 HTTP/1.1\r\n",
                        "GET /secure HTTP/1.1\r\n",
                        "GET http://localhost:2/final HTTP/1.1\r\n",
                    ])
                    .enumerate()
                {
                    let headers: HashMap<_, _> = request
                        .lines()
                        .skip(1)
                        .filter_map(|line| line.split_once(':'))
                        .map(|(name, value)| (name.to_ascii_lowercase(), value.trim()))
                        .collect();
                    assert!(request.starts_with(target), "{request}");
                    assert_eq!(headers.get("proxy-authorization").copied(), (index != 2).then_some(expected_auth.as_str()), "{request}");
                    assert_eq!(headers.get("authorization").copied(), (index == 0).then_some("Bearer origin-19"), "{request}");
                    assert_eq!(headers.get("cookie").copied(), (index == 0).then_some("session=origin-23"), "{request}");
                }

                assert_eq!(response.status.as_u16(), 200);
                assert_eq!(response.headers, HashMap::new());
                assert_eq!(response.body, Bytes::from_static(b"accepted"));
            })
            .await
            .unwrap();
        });
}

#[cfg(all(unix, not(target_os = "android"), not(target_vendor = "apple")))]
#[rstest]
fn http2_initial_window_follows_environment() {
    let Some(config) = tls_server_config() else {
        for (value, expected) in [
            (None, "windows=16777216/33554432"),
            (Some("false"), "windows=16777216/33554432"),
            (Some("true"), "windows=65535/65535"),
            (
                Some("1"),
                "error=Failed to build HTTP client: NAUTILUS_HTTP2_ADAPTIVE_WINDOW must be 'true' or 'false', was '1'",
            ),
        ] {
            let stdout = run_tls_child(
                "http::tests::http2_initial_window_follows_environment",
                &[("NAUTILUS_HTTP2_ADAPTIVE_WINDOW", value)],
            );
            assert!(stdout.contains(expected), "{value:?}: {stdout}");
        }

        return;
    };

    tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap()
        .block_on(async {
            let client = match HttpClient::builder()
                .use_system_proxy(false)
                .timeout_secs(3)
                .build()
            {
                Ok(client) => client,
                Err(e) => {
                    println!("error={e}");
                    return;
                }
            };

            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let port = listener.local_addr().unwrap().port();
            let url = format!("https://localhost:{port}/");

            let mut task = tokio::spawn(async move { send(&client, Method::GET, url, None).await });

            let (stream, _) = tokio::select! {
                accepted = listener.accept() => accepted.unwrap(),
                result = &mut task => panic!("request ended before connecting: {result:?}"),
            };
            let mut stream = tokio_rustls::TlsAcceptor::from(config)
                .accept(stream)
                .await
                .unwrap();

            let mut preface = [0; 24];
            stream.read_exact(&mut preface).await.unwrap();
            assert_eq!(&preface, b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n");

            // Both windows start at 65,535 bytes until SETTINGS or WINDOW_UPDATE frames change them
            let mut stream_window = 65_535;
            let mut connection_window = 65_535;

            loop {
                let mut header = [0; 9];
                stream.read_exact(&mut header).await.unwrap();
                let length = u32::from_be_bytes([0, header[0], header[1], header[2]]);
                let id = u32::from_be_bytes([header[5], header[6], header[7], header[8]]);
                let mut payload = vec![0; length as usize];
                stream.read_exact(&mut payload).await.unwrap();

                match header[3] {
                    0x1 => break,
                    0x4 => {
                        for setting in payload.as_chunks::<6>().0 {
                            if let [0, 0x4, a, b, c, d] = *setting {
                                stream_window = u32::from_be_bytes([a, b, c, d]);
                            }
                        }
                    }
                    0x8 if id == 0 => {
                        connection_window +=
                            u32::from_be_bytes([payload[0], payload[1], payload[2], payload[3]]);
                    }
                    _ => {}
                }
            }

            println!("windows={stream_window}/{connection_window}");
            task.abort();
        });
}

#[cfg(all(unix, not(target_os = "android"), not(target_vendor = "apple")))]
#[rstest]
fn platform_roots_load_once_per_process() {
    let Some(config) = tls_server_config() else {
        // Cargo exports the system `SSL_CERT_DIR`, which would still supply roots without the file
        run_tls_child(
            "http::tests::platform_roots_load_once_per_process",
            &[("SSL_CERT_DIR", None)],
        );
        return;
    };

    let roots = std::path::PathBuf::from(std::env::var_os("SSL_CERT_FILE").unwrap());
    let parked = roots.with_extension("parked");

    tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap()
        .block_on(async {
            std::fs::rename(&roots, &parked).unwrap();
            let error = HttpClient::builder()
                .use_system_proxy(false)
                .build()
                .unwrap_err();

            std::fs::rename(&parked, &roots).unwrap();
            HttpClient::builder()
                .use_system_proxy(false)
                .build()
                .unwrap();

            std::fs::remove_file(&roots).unwrap();
            let client = HttpClient::builder()
                .use_system_proxy(false)
                .timeout_secs(3)
                .build()
                .unwrap();

            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let port = listener.local_addr().unwrap().port();
            let acceptor = tokio_rustls::TlsAcceptor::from(config);

            let peer = tokio::spawn(async move {
                let (stream, _) = listener.accept().await.unwrap();
                let stream = acceptor.accept(stream).await.unwrap();
                let mut connection = h2::server::handshake(stream).await.unwrap();
                while let Some(request) = connection.accept().await {
                    let (_, mut respond) = request.unwrap();
                    respond
                        .send_response(http::Response::new(()), true)
                        .unwrap();
                }
            });

            let response = send(
                &client,
                Method::GET,
                format!("https://localhost:{port}/"),
                None,
            )
            .await
            .unwrap();
            peer.abort();

            assert_eq!(
                error.to_string(),
                "Failed to build HTTP client: unexpected error: No CA certificates were loaded from the system"
            );
            assert_eq!(response.status.as_u16(), 200);
            assert_eq!(response.body, Bytes::new());
        });
}

// SSL_CERT_FILE supplies isolated trust on the Unix verifier, not native Apple/Windows stores
#[cfg(all(unix, not(target_os = "android"), not(target_vendor = "apple")))]
const TLS_CHILD: &str = "NAUTILUS_HTTP_TLS_PARITY_CHILD";

#[cfg(all(unix, not(target_os = "android"), not(target_vendor = "apple")))]
fn run_tls_child(test: &str, envs: &[(&str, Option<&str>)]) -> String {
    let directory = tempfile::tempdir().unwrap();
    let key = rcgen::KeyPair::generate().unwrap();
    let cert = rcgen::CertificateParams::new(vec!["localhost".into()])
        .unwrap()
        .self_signed(&key)
        .unwrap();
    std::fs::write(directory.path().join("cert.pem"), cert.pem()).unwrap();
    std::fs::write(directory.path().join("cert.der"), cert.der()).unwrap();
    std::fs::write(directory.path().join("key.der"), key.serialize_der()).unwrap();
    let mut command = std::process::Command::new(std::env::current_exe().unwrap());
    command
        .args(["--exact", test, "--nocapture"])
        .env(TLS_CHILD, directory.path())
        .env("SSL_CERT_FILE", directory.path().join("cert.pem"));

    for (key, value) in envs {
        match value {
            Some(value) => command.env(key, value),
            None => command.env_remove(key),
        };
    }

    let output = command.output().unwrap();
    let stdout = String::from_utf8_lossy(&output.stdout).into_owned();
    assert!(
        output.status.success(),
        "{stdout}\n{}",
        String::from_utf8_lossy(&output.stderr)
    );
    assert!(stdout.contains("1 passed"));
    stdout
}

#[cfg(all(unix, not(target_os = "android"), not(target_vendor = "apple")))]
fn tls_server_config() -> Option<Arc<rustls::ServerConfig>> {
    let directory = std::path::PathBuf::from(std::env::var_os(TLS_CHILD)?);
    nautilus_cryptography::providers::install_cryptographic_provider();
    let cert =
        rustls::pki_types::CertificateDer::from(std::fs::read(directory.join("cert.der")).unwrap());
    let key = rustls::pki_types::PrivatePkcs8KeyDer::from(
        std::fs::read(directory.join("key.der")).unwrap(),
    );
    let mut config = rustls::ServerConfig::builder()
        .with_no_client_auth()
        .with_single_cert(vec![cert], key.into())
        .unwrap();
    config.alpn_protocols = vec![b"h2".to_vec()];
    Some(Arc::new(config))
}

async fn send(
    client: &HttpClient,
    method: Method,
    url: String,
    body: Option<Vec<u8>>,
) -> Result<HttpResponse, HttpClientError> {
    client
        .request(method, url, None, None, body, None, None)
        .await
}

async fn read_headers(stream: &mut (impl tokio::io::AsyncRead + Unpin)) -> String {
    let mut bytes = Vec::new();
    while !bytes.ends_with(b"\r\n\r\n") {
        bytes.push(stream.read_u8().await.unwrap());
    }
    String::from_utf8(bytes).unwrap()
}