ntex 4.1.0

#![cfg(feature = "rustls")]
//! Same as `http_rustls`, but the client also uses rustls
use std::sync::{Arc, Mutex, atomic::AtomicBool, atomic::AtomicUsize, atomic::Ordering};
use std::{future::ready, io, net::SocketAddr};

use futures_util::stream::{Stream, StreamExt, once};

use ntex::codec::BytesCodec;
use ntex::http::error::PayloadError;
use ntex::http::header::{self, HeaderName, HeaderValue};
use ntex::http::test::{self, server as test_server};
use ntex::http::{
    self, HttpService, Method, Request, Response, StatusCode, Version, body, h1,
    h2::Http2ServiceConfig,
};
use ntex::io::{Filter, IoConfig};
use ntex::service::cfg::SharedCfg;
use ntex::time::{Millis, Seconds, sleep, timeout};
use ntex::util::{Bytes, BytesMut, Either, select};
use ntex::ws::{self, WsClient, WsClientConfig, WsConnection, handshake_response};
use ntex::{channel::oneshot, client, rt, web::error::InternalError};
use ntex_tls::TlsConfig;

mod rustls_utils;
use rustls_utils::{tls_acceptor, tls_connector};

/// rustls client with the same settings as the test server's client
struct RustlsClient {
    addr: SocketAddr,
    cfg: SharedCfg,
    client: client::Client,
}

impl RustlsClient {
    fn new(srv: &test::TestServer) -> Self {
        let addr = srv.addr();
        let cfg = SharedCfg::new("TEST-CLIENT")
            .add(IoConfig::new().set_connect_timeout(Millis(90_000)))
            .add(TlsConfig::new().set_handshake_timeout(Seconds(90)))
            .add(
                client::ClientConfig::new()
                    .set_response_timeout(Seconds(30))
                    .set_response_payload_timeout(Seconds(30)),
            )
            .add(
                Http2ServiceConfig::new()
                    .set_max_header_list_size(256 * 1024)
                    .set_max_header_continuation_frames(96),
            )
            .add(
                WsClientConfig::new()
                    .set_address(addr)
                    .set_handshake_timeout(Seconds(30)),
            )
            .build();

        let mut config = tls_connector();
        config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
        let client = client::Client::builder().rustls(config).build(cfg.clone());

        RustlsClient { addr, cfg, client }
    }

    fn surl(&self, path: &str) -> String {
        format!("https://localhost:{}{}", self.addr.port(), path)
    }

    fn srequest<S: AsRef<str>>(&self, method: Method, path: S) -> client::ClientRequest {
        self.client
            .request(method, self.surl(path.as_ref()).as_str())
    }

    async fn wss(&self) -> WsConnection<impl Filter> {
        let mut config = tls_connector();
        config.alpn_protocols = vec![b"http/1.1".to_vec()];

        WsClient::new(self.surl("/"), &self.cfg)
            .rustls(Arc::new(config))
            .connect()
            .await
            .unwrap()
    }
}

async fn load_body<S>(stream: S) -> Result<BytesMut, PayloadError>
where
    S: Stream<Item = Result<Bytes, PayloadError>>,
{
    let body = stream
        .map(|res| if let Ok(chunk) = res { chunk } else { panic!() })
        .fold(BytesMut::new(), async move |mut body, chunk| {
            body.extend_from_slice(&chunk);
            body
        })
        .await;

    Ok(body)
}

#[ntex::test]
async fn test_h2() -> io::Result<()> {
    let srv = test_server(async move |_| {
        http::rustls(
            tls_acceptor(),
            http::ALPN_PROTO_H2,
            HttpService::h2(async |_| Ok::<_, io::Error>(Response::Ok().build())),
        )
    });
    let cl = RustlsClient::new(&srv);

    let response = cl.srequest(Method::GET, "/").send().await.unwrap();
    assert!(response.status().is_success());
    Ok(())
}

#[ntex::test]
async fn test_h1() -> io::Result<()> {
    let srv = test_server(async move |_| {
        http::rustls(
            tls_acceptor(),
            http::ALPN_PROTO_H1,
            HttpService::h1(async |_| Ok::<_, io::Error>(Response::Ok().build())),
        )
    });
    let cl = RustlsClient::new(&srv);

    let response = cl.srequest(Method::GET, "/").send().await.unwrap();
    assert!(response.status().is_success());
    Ok(())
}

#[ntex::test]
async fn test_h2_1() -> io::Result<()> {
    let srv = test_server(async move |_| {
        http::rustls(
            tls_acceptor(),
            http::ALPN_PROTOS,
            HttpService::new(async |req: Request| {
                assert!(req.peer_addr().is_some());
                assert_eq!(req.version(), Version::HTTP_2);
                Ok::<_, io::Error>(Response::Ok().build())
            }),
        )
    });
    let cl = RustlsClient::new(&srv);

    let response = cl.srequest(Method::GET, "/").send().await.unwrap();
    assert!(response.status().is_success());
    Ok(())
}

#[ntex::test]
async fn test_h2_body() -> io::Result<()> {
    let data = "HELLOWORLD".to_owned().repeat(64 * 1024);
    let srv = test_server(async move |_| {
        http::rustls(
            tls_acceptor(),
            http::ALPN_PROTO_H2,
            HttpService::h2(async move |mut req: Request| {
                let body = load_body(req.take_payload())
                    .await
                    .map_err(io::Error::other)?;
                Ok::<_, io::Error>(Response::Ok().body(body))
            }),
        )
    });
    let cl = RustlsClient::new(&srv);

    let response = cl
        .srequest(Method::GET, "/")
        .send_body(data.clone())
        .await
        .unwrap();
    assert!(response.status().is_success());

    let body = srv.load_body(response).await.unwrap();
    assert_eq!(&body, data.as_bytes());
    Ok(())
}

#[ntex::test]
async fn test_h2_content_length() {
    let srv = test_server(async move |_| {
        http::rustls(
            tls_acceptor(),
            http::ALPN_PROTO_H2,
            HttpService::h2(async move |req: Request| {
                let indx: usize = req.uri().path().as_str()[1..].parse().unwrap();
                let statuses = [
                    StatusCode::NO_CONTENT,
                    // h2 lib does not accept hangs on this statuses
                    //StatusCode::CONTINUE,
                    //StatusCode::SWITCHING_PROTOCOLS,
                    //StatusCode::PROCESSING,
                    StatusCode::OK,
                    StatusCode::NOT_FOUND,
                ];
                Ok::<_, io::Error>(Response::new(statuses[indx]))
            }),
        )
    });
    let cl = RustlsClient::new(&srv);

    let header = HeaderName::from_static("content-length");
    let value = HeaderValue::from_static("0");

    {
        for i in 0..1 {
            let req = cl.srequest(Method::GET, format!("/{i}")).send();
            let response = req.await.unwrap();
            assert_eq!(response.headers().get(&header), None);

            let req = cl.srequest(Method::HEAD, format!("/{i}")).send();
            let response = req.await.unwrap();
            assert_eq!(response.headers().get(&header), None);
        }

        for i in 1..3 {
            let req = cl.srequest(Method::GET, format!("/{i}")).send();
            let response = req.await.unwrap();
            assert_eq!(response.headers().get(&header), Some(&value));
        }
    }
}

#[ntex::test]
async fn test_h2_headers() {
    let data = STR.repeat(10);
    let data2 = data.clone();

    let srv = test_server(async move |_| {
        let data = data.clone();
        http::rustls(
            tls_acceptor(),
            http::ALPN_PROTO_H2,
            HttpService::h2(async move |_| {
                let mut builder = Response::Ok();
                for idx in 0..90 {
                    builder.header(
                    format!("X-TEST-{idx}").as_str(),
                    "TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
                        TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
                        TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
                        TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
                        TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
                        TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
                        TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
                        TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
                        TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
                        TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
                        TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
                        TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
                        TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST ",
                );
                }
                Ok::<_, io::Error>(builder.body(data.clone()))
            }),
        )
    });
    let cl = RustlsClient::new(&srv);

    let response = cl.srequest(Method::GET, "/").send().await.unwrap();
    assert!(response.status().is_success());

    // read response
    let bytes = srv.load_body(response).await.unwrap();
    assert_eq!(bytes, Bytes::from(data2));
}

const STR: &str = "Hello World Hello World Hello World Hello World Hello World \
                   Hello World Hello World Hello World Hello World Hello World \
                   Hello World Hello World Hello World Hello World Hello World \
                   Hello World Hello World Hello World Hello World Hello World \
                   Hello World Hello World Hello World Hello World Hello World \
                   Hello World Hello World Hello World Hello World Hello World \
                   Hello World Hello World Hello World Hello World Hello World \
                   Hello World Hello World Hello World Hello World Hello World \
                   Hello World Hello World Hello World Hello World Hello World \
                   Hello World Hello World Hello World Hello World Hello World \
                   Hello World Hello World Hello World Hello World Hello World \
                   Hello World Hello World Hello World Hello World Hello World \
                   Hello World Hello World Hello World Hello World Hello World \
                   Hello World Hello World Hello World Hello World Hello World \
                   Hello World Hello World Hello World Hello World Hello World \
                   Hello World Hello World Hello World Hello World Hello World \
                   Hello World Hello World Hello World Hello World Hello World \
                   Hello World Hello World Hello World Hello World Hello World \
                   Hello World Hello World Hello World Hello World Hello World \
                   Hello World Hello World Hello World Hello World Hello World \
                   Hello World Hello World Hello World Hello World Hello World";

#[ntex::test]
async fn test_h2_body2() {
    let srv = test_server(async move |_| {
        http::rustls(
            tls_acceptor(),
            http::ALPN_PROTO_H2,
            HttpService::h2(async |_| Ok::<_, io::Error>(Response::Ok().body(STR))),
        )
    });
    let cl = RustlsClient::new(&srv);

    let response = cl.srequest(Method::GET, "/").send().await.unwrap();
    assert!(response.status().is_success());

    // read response
    let bytes = srv.load_body(response).await.unwrap();
    assert_eq!(bytes, Bytes::from_static(STR.as_ref()));
}

#[ntex::test]
async fn test_h2_head_empty() {
    let srv = test_server(async move |_| {
        http::rustls(
            tls_acceptor(),
            http::ALPN_PROTO_H2,
            HttpService::h2(async |_| Ok::<_, io::Error>(Response::Ok().body(STR))),
        )
    });
    let cl = RustlsClient::new(&srv);

    let response = cl.srequest(Method::HEAD, "/").send().await.unwrap();
    assert!(response.status().is_success());
    assert_eq!(response.version(), Version::HTTP_2);

    {
        let len = response.headers().get(header::CONTENT_LENGTH).unwrap();
        assert_eq!(format!("{}", STR.len()), len.to_str().unwrap());
    }

    // read response
    let bytes = srv.load_body(response).await.unwrap();
    assert!(bytes.is_empty());
}

#[ntex::test]
async fn test_h2_head_binary() {
    let srv = test_server(async move |_| {
        http::rustls(
            tls_acceptor(),
            http::ALPN_PROTO_H2,
            HttpService::h2(async |_| {
                Ok::<_, io::Error>(Response::Ok().content_length(STR.len() as u64).body(STR))
            }),
        )
    });
    let cl = RustlsClient::new(&srv);

    let response = cl.srequest(Method::HEAD, "/").send().await.unwrap();
    assert!(response.status().is_success());

    {
        let len = response.headers().get(header::CONTENT_LENGTH).unwrap();
        assert_eq!(format!("{}", STR.len()), len.to_str().unwrap());
    }

    // read response
    let bytes = srv.load_body(response).await.unwrap();
    assert!(bytes.is_empty());
}

/// Server must send content-length, but no payload
#[ntex::test]
async fn test_h2_head_binary2() {
    let srv = test_server(async move |_| {
        http::rustls(
            tls_acceptor(),
            http::ALPN_PROTO_H2,
            HttpService::h2(async |_| Ok::<_, io::Error>(Response::Ok().body(STR))),
        )
    });
    let cl = RustlsClient::new(&srv);

    let response = cl.srequest(Method::HEAD, "/").send().await.unwrap();
    assert!(response.status().is_success());

    {
        let len = response.headers().get(header::CONTENT_LENGTH).unwrap();
        assert_eq!(format!("{}", STR.len()), len.to_str().unwrap());
    }
}

#[ntex::test]
async fn test_h2_body_length() {
    let srv = test_server(async move |_| {
        http::rustls(
            tls_acceptor(),
            http::ALPN_PROTO_H2,
            HttpService::h2(async |_| {
                let body = once(ready(Ok(Bytes::from_static(STR.as_ref()))));
                Ok::<_, io::Error>(
                    Response::Ok().body(body::SizedStream::new(STR.len() as u64, body)),
                )
            }),
        )
    });
    let cl = RustlsClient::new(&srv);

    let response = cl.srequest(Method::GET, "/").send().await.unwrap();
    assert!(response.status().is_success());

    // read response
    let bytes = srv.load_body(response).await.unwrap();
    assert_eq!(bytes, Bytes::from_static(STR.as_ref()));
}

#[ntex::test]
async fn test_h2_body_chunked_explicit() {
    let srv = test_server(async move |_| {
        http::rustls(
            tls_acceptor(),
            http::ALPN_PROTO_H2,
            HttpService::h2(async |_| {
                let body = once(ready(Ok::<_, io::Error>(Bytes::from_static(STR.as_ref()))));
                Ok::<_, io::Error>(
                    Response::Ok()
                        .header(header::TRANSFER_ENCODING, "chunked")
                        .streaming(body),
                )
            }),
        )
    });
    let cl = RustlsClient::new(&srv);

    let response = cl.srequest(Method::GET, "/").send().await.unwrap();
    assert!(response.status().is_success());
    assert!(!response.headers().contains_key(header::TRANSFER_ENCODING));

    // read response
    let bytes = srv.load_body(response).await.unwrap();

    // decode
    assert_eq!(bytes, Bytes::from_static(STR.as_ref()));
}

#[ntex::test]
async fn test_h2_response_http_error_handling() {
    let srv = test_server(async move |_| {
        http::rustls(
            tls_acceptor(),
            http::ALPN_PROTO_H2,
            HttpService::h2(async |_| {
                let broken_header = Bytes::from_static(b"\0\0\0");
                Ok::<_, io::Error>(
                    Response::Ok()
                        .header(header::CONTENT_TYPE, &broken_header[..])
                        .body(STR),
                )
            }),
        )
    });
    let cl = RustlsClient::new(&srv);

    let response = cl.srequest(Method::GET, "/").send().await.unwrap();
    assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);

    // read response
    let bytes = srv.load_body(response).await.unwrap();
    assert_eq!(bytes, Bytes::from_static(b"Invalid HTTP header value"));
}

#[ntex::test]
async fn test_h2_service_error() {
    let srv = test_server(async move |_| {
        http::rustls(
            tls_acceptor(),
            http::ALPN_PROTO_H2,
            HttpService::h2(async |_| {
                Err::<Response, _>(InternalError::default("error", StatusCode::BAD_REQUEST))
            }),
        )
    });
    let cl = RustlsClient::new(&srv);

    let response = cl.srequest(Method::GET, "/").send().await.unwrap();
    assert_eq!(response.status(), StatusCode::BAD_REQUEST);

    // read response
    let bytes = srv.load_body(response).await.unwrap();
    assert_eq!(bytes, Bytes::from_static(b"error"));
}

struct SetOnDrop(Arc<AtomicUsize>, Arc<Mutex<Option<::oneshot::Sender<()>>>>);

impl Drop for SetOnDrop {
    fn drop(&mut self) {
        self.0.fetch_add(1, Ordering::Relaxed);
        let _ = self.1.lock().unwrap().take().unwrap().send(());
    }
}

#[ntex::test]
async fn test_h2_client_drop() -> io::Result<()> {
    let count = Arc::new(AtomicUsize::new(0));
    let count2 = count.clone();
    let started = Arc::new(AtomicBool::new(false));
    let started2 = started.clone();
    let (tx, rx) = ::oneshot::async_channel();
    let tx = Arc::new(Mutex::new(Some(tx)));

    let srv = test_server(async move |_| {
        let tx = tx.clone();
        let count = count2.clone();
        let started = started2.clone();
        http::rustls(
            tls_acceptor(),
            http::ALPN_PROTO_H2,
            HttpService::h2(async move |req: Request| {
                let st = SetOnDrop(count.clone(), tx.clone());
                started.store(true, Ordering::Relaxed);

                assert!(req.peer_addr().is_some());
                assert_eq!(req.version(), Version::HTTP_2);
                sleep(Seconds(30)).await;
                drop(st);
                Ok::<_, io::Error>(Response::Ok().build())
            }),
        )
    });
    let cl = RustlsClient::new(&srv);

    // drop the request once the handler is running
    let wait_started = async {
        while !started.load(Ordering::Relaxed) {
            sleep(Millis(10)).await;
        }
    };
    let result = timeout(
        Millis(2500),
        select(cl.srequest(Method::GET, "/").send(), wait_started),
    )
    .await;
    assert!(matches!(result, Ok(Either::Right(()))));
    let _ = timeout(Millis(1500), rx).await;
    assert_eq!(count.load(Ordering::Relaxed), 1);
    Ok(())
}

#[ntex::test]
async fn test_ssl_handshake_timeout() {
    use std::io::Read;

    let srv = test::server_with_config(
        async move |_| {
            http::rustls(
                tls_acceptor(),
                http::ALPN_PROTO_H2,
                HttpService::h2(async |_| Ok::<_, io::Error>(Response::Ok().build())),
            )
        },
        SharedCfg::new("SVC").add(TlsConfig::new().set_handshake_timeout(Seconds(1))),
    );

    let mut stream = std::net::TcpStream::connect(srv.addr()).unwrap();
    let mut data = String::new();
    let _ = stream.read_to_string(&mut data);
    assert!(data.is_empty());
}

#[ntex::test]
async fn test_ws_transport() {
    let srv = test_server(async |_| {
        http::rustls(
            tls_acceptor(),
            http::ALPN_PROTOS,
            HttpService::new(async |_| Ok::<_, io::Error>(Response::NotFound())).h1_control(
                async move |req: h1::Control<_, _>| {
                    let ack = if let h1::Control::Upgrade(upg) = req {
                        let (ack, io, req, codec) = upg.handle();

                        // send handshake respone
                        let res = handshake_response(req.head()).build();
                        io.encode(
                            h1::Message::Item((res.drop_body(), body::BodySize::None)),
                            &codec,
                        )
                        .unwrap();

                        // start websocket service
                        let io = ws::WsTransport::create(io, ws::Codec::default());
                        while let Some(item) =
                            io.recv(&BytesCodec).await.map_err(|e| e.into_inner())?
                        {
                            io.send(item, &BytesCodec).await.unwrap()
                        }

                        ack
                    } else {
                        req.ack()
                    };
                    Ok::<_, io::Error>(ack)
                },
            ),
        )
    });
    let cl = RustlsClient::new(&srv);

    let io = cl.wss().await.into_inner().0;
    let codec = ws::Codec::default().set_client_mode();

    io.send(ws::Message::Binary(Bytes::from_static(b"text")), &codec)
        .await
        .unwrap();

    let item = io.recv(&codec).await.unwrap().unwrap();
    assert_eq!(item, ws::Frame::Binary(Bytes::from_static(b"text")));

    // the transport echoes the close code
    io.send(ws::Message::Close(Some(ws::CloseCode::Away.into())), &codec)
        .await
        .unwrap();
    let item = io.recv(&codec).await.unwrap().unwrap();
    assert_eq!(item, ws::Frame::Close(Some(ws::CloseCode::Away.into())));
}

#[ntex::test]
async fn test_h2_not_graceful_shutdown() -> io::Result<()> {
    let count = Arc::new(AtomicUsize::new(0));
    let count2 = count.clone();
    let (tx, rx) = ::oneshot::channel();
    let tx = Arc::new(Mutex::new(Some(tx)));

    let srv = test_server(async move |_| {
        let tx = tx.clone();
        let count = count2.clone();
        http::rustls(
            tls_acceptor(),
            http::ALPN_PROTO_H2,
            HttpService::h2(async move |_| {
                let count = count.clone();
                count.fetch_add(1, Ordering::Relaxed);
                if count.load(Ordering::Relaxed) == 2 {
                    let _ = tx.lock().unwrap().take().unwrap().send(());
                }
                sleep(Millis(1000)).await;
                count.fetch_sub(1, Ordering::Relaxed);
                Ok::<_, io::Error>(Response::Ok().build())
            }),
        )
    });
    let cl = RustlsClient::new(&srv);

    let req = cl.srequest(Method::GET, "/");
    rt::spawn(async move {
        assert!(matches!(
            req.send().await.err().unwrap().into_error(),
            client::error::ClientError::H2 { .. }
        ));
        sleep(Millis(100000)).await;
    });
    let req = cl.srequest(Method::GET, "/");
    rt::spawn(async move {
        assert!(matches!(
            req.send().await.err().unwrap().into_error(),
            client::error::ClientError::H2 { .. }
        ));
        sleep(Millis(100000)).await;
    });
    let _ = rx.await;
    assert_eq!(count.load(Ordering::Relaxed), 2);

    let (tx, rx) = oneshot::channel();
    rt::spawn(async move {
        srv.stop(false).await;
        let _ = tx.send(());
    });

    let _ = rx.await;
    assert_eq!(count.load(Ordering::Relaxed), 2);
    Ok(())
}