567 lines
19 KiB
Rust
567 lines
19 KiB
Rust
use std::future::{Future, poll_fn};
|
|
use std::time::Duration;
|
|
|
|
use dever_runtime::{
|
|
async_stream::AsyncStream, bytes::Bytes, channel::Channel, http, net, task, tls,
|
|
};
|
|
use tokio::sync::{mpsc, oneshot};
|
|
|
|
const ROOT: &[u8] = include_bytes!("../fixtures/tls/root.pem");
|
|
const CERT: &[u8] = include_bytes!("../fixtures/tls/server.pem");
|
|
const KEY: &[u8] = include_bytes!("../fixtures/tls/server-key.pem");
|
|
|
|
fn bytes(value: &[u8]) -> Bytes {
|
|
Bytes::new(value.to_vec())
|
|
}
|
|
|
|
fn limits(connections: i64, streams: i64) -> http::Limits {
|
|
http::Limits {
|
|
header_bytes: 8192,
|
|
body_bytes: 65536,
|
|
timeout_ms: 1000,
|
|
connections,
|
|
http2: Some(http::Http2Limits {
|
|
streams,
|
|
stream_window_bytes: 65535,
|
|
connection_window_bytes: 65535,
|
|
}),
|
|
}
|
|
}
|
|
|
|
fn pool_limits() -> http::PoolLimits {
|
|
http::PoolLimits {
|
|
idle_ms: 1000,
|
|
chunk_bytes: 4,
|
|
read_ms: 500,
|
|
}
|
|
}
|
|
|
|
fn request(target: &str) -> http::Request {
|
|
http::Request {
|
|
method: "GET".into(),
|
|
target: target.into(),
|
|
headers: vec![],
|
|
body: bytes(b""),
|
|
}
|
|
}
|
|
|
|
fn trust() -> tls::ClientTls {
|
|
tls::client(&bytes(ROOT)).unwrap()
|
|
}
|
|
|
|
fn identity() -> tls::ServerTls {
|
|
tls::server(&bytes(CERT), &bytes(KEY)).unwrap()
|
|
}
|
|
|
|
fn run(future: impl Future<Output = Result<(), String>>) {
|
|
task::run_entry_with(
|
|
task::RuntimeConfig {
|
|
worker_threads: 1,
|
|
max_blocking_threads: 2,
|
|
task_capacity: 64,
|
|
},
|
|
async {
|
|
tokio::time::timeout(Duration::from_secs(8), future)
|
|
.await
|
|
.map_err(|_| "HTTP/2 client test deadline exceeded".to_owned())?
|
|
},
|
|
)
|
|
.unwrap();
|
|
}
|
|
|
|
fn h2_error(error: impl std::fmt::Display) -> String {
|
|
error.to_string()
|
|
}
|
|
|
|
async fn next_request(
|
|
requests: &mut mpsc::Receiver<(
|
|
hyper::Request<h2::RecvStream>,
|
|
h2::server::SendResponse<bytes::Bytes>,
|
|
)>,
|
|
) -> Result<
|
|
(
|
|
hyper::Request<h2::RecvStream>,
|
|
h2::server::SendResponse<bytes::Bytes>,
|
|
),
|
|
String,
|
|
> {
|
|
requests
|
|
.recv()
|
|
.await
|
|
.ok_or("HTTP/2 connection closed before request".into())
|
|
}
|
|
|
|
fn response() -> hyper::Response<()> {
|
|
hyper::Response::builder().status(200).body(()).unwrap()
|
|
}
|
|
|
|
async fn start_peer(
|
|
socket: tokio::net::TcpStream,
|
|
) -> Result<
|
|
(
|
|
mpsc::Receiver<(
|
|
hyper::Request<h2::RecvStream>,
|
|
h2::server::SendResponse<bytes::Bytes>,
|
|
)>,
|
|
oneshot::Sender<oneshot::Sender<()>>,
|
|
task::Task<()>,
|
|
),
|
|
String,
|
|
> {
|
|
let mut connection = h2::server::handshake(socket).await.map_err(h2_error)?;
|
|
let (requests, receiver) = mpsc::channel(4);
|
|
let (shutdown, mut shutdown_request) = oneshot::channel::<oneshot::Sender<()>>();
|
|
let driver = task::run(async move {
|
|
let mut graceful = false;
|
|
loop {
|
|
let request = if graceful {
|
|
connection.accept().await
|
|
} else {
|
|
tokio::select! {
|
|
request = connection.accept() => request,
|
|
command = &mut shutdown_request => {
|
|
if let Ok(acknowledge) = command {
|
|
connection.graceful_shutdown();
|
|
graceful = true;
|
|
let _ = acknowledge.send(());
|
|
continue;
|
|
}
|
|
connection.accept().await
|
|
}
|
|
}
|
|
};
|
|
let Some(request) = request else {
|
|
return Ok(());
|
|
};
|
|
requests
|
|
.send(request.map_err(h2_error)?)
|
|
.await
|
|
.map_err(|_| "HTTP/2 peer request receiver closed".to_owned())?;
|
|
}
|
|
})
|
|
.await?;
|
|
Ok((receiver, shutdown, driver))
|
|
}
|
|
|
|
#[test]
|
|
fn one_physical_connection_multiplexes_upload_and_streaming_download() {
|
|
run(async {
|
|
let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
|
|
.await
|
|
.map_err(h2_error)?;
|
|
let port = i64::from(listener.local_addr().map_err(h2_error)?.port());
|
|
let peer = task::run(async move {
|
|
let (socket, _) = listener.accept().await.map_err(h2_error)?;
|
|
let (mut requests, _shutdown, driver) = start_peer(socket).await?;
|
|
|
|
let (first, mut first_reply) = next_request(&mut requests).await?;
|
|
assert_eq!(first.method(), hyper::Method::POST);
|
|
assert_eq!(first.uri().scheme_str(), Some("http"));
|
|
let authority = format!("localhost:{port}");
|
|
assert_eq!(
|
|
first.uri().authority().map(|value| value.as_str()),
|
|
Some(authority.as_str())
|
|
);
|
|
assert_eq!(first.uri().path(), "/upload");
|
|
let mut upload = first.into_body();
|
|
let mut uploaded = Vec::new();
|
|
while let Some(chunk) = upload.data().await {
|
|
let chunk = chunk.map_err(h2_error)?;
|
|
uploaded.extend_from_slice(&chunk);
|
|
}
|
|
assert_eq!(uploaded, b"abcdefgh");
|
|
let mut first_body = first_reply
|
|
.send_response(response(), false)
|
|
.map_err(h2_error)?;
|
|
first_body
|
|
.send_data(bytes::Bytes::from_static(b"first"), false)
|
|
.map_err(h2_error)?;
|
|
|
|
let (second, mut second_reply) = next_request(&mut requests).await?;
|
|
assert_eq!(second.uri().path(), "/second");
|
|
let mut second_body = second_reply
|
|
.send_response(response(), false)
|
|
.map_err(h2_error)?;
|
|
second_body
|
|
.send_data(bytes::Bytes::from_static(b"second"), true)
|
|
.map_err(h2_error)?;
|
|
first_body
|
|
.send_data(bytes::Bytes::from_static(b"last"), true)
|
|
.map_err(h2_error)?;
|
|
|
|
task::wait(driver).await?;
|
|
Ok(())
|
|
})
|
|
.await?;
|
|
|
|
let client = http::client(
|
|
&format!("http://localhost:{port}"),
|
|
tls::system(),
|
|
limits(1, 2),
|
|
pool_limits(),
|
|
)
|
|
.await?;
|
|
let first = http::open_stream(
|
|
&client,
|
|
http::StreamRequest {
|
|
method: "POST".into(),
|
|
target: "/upload".into(),
|
|
headers: vec![],
|
|
body: AsyncStream::from_values([Ok(bytes(b"abcd")), Ok(bytes(b"efgh"))]),
|
|
},
|
|
)
|
|
.await?;
|
|
assert_eq!(first.body.pull().await.unwrap()?.values(), b"firs");
|
|
assert_eq!(first.body.pull().await.unwrap()?.values(), b"t");
|
|
assert_eq!(
|
|
http::request(&client, request("/second"))
|
|
.await?
|
|
.body
|
|
.values(),
|
|
b"second"
|
|
);
|
|
let mut tail = Vec::new();
|
|
while let Some(chunk) = first.body.pull().await {
|
|
tail.extend_from_slice(chunk?.values());
|
|
}
|
|
assert_eq!(tail, b"last");
|
|
http::close_client(&client).await?;
|
|
task::wait(peer).await?;
|
|
Ok(())
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn canceling_one_response_resets_only_its_stream() {
|
|
run(async {
|
|
let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
|
|
.await
|
|
.map_err(h2_error)?;
|
|
let port = i64::from(listener.local_addr().map_err(h2_error)?.port());
|
|
let peer = task::run(async move {
|
|
let (socket, _) = listener.accept().await.map_err(h2_error)?;
|
|
let (mut requests, _shutdown, driver) = start_peer(socket).await?;
|
|
let (_, mut first_reply) = next_request(&mut requests).await?;
|
|
let mut first_body = first_reply
|
|
.send_response(response(), false)
|
|
.map_err(h2_error)?;
|
|
first_body
|
|
.send_data(bytes::Bytes::from_static(b"open"), false)
|
|
.map_err(h2_error)?;
|
|
|
|
let (_, mut neighbor_reply) = next_request(&mut requests).await?;
|
|
let mut neighbor_body = neighbor_reply
|
|
.send_response(response(), false)
|
|
.map_err(h2_error)?;
|
|
neighbor_body
|
|
.send_data(bytes::Bytes::from_static(b"neighbor"), true)
|
|
.map_err(h2_error)?;
|
|
|
|
let reason = poll_fn(|context| first_body.poll_reset(context))
|
|
.await
|
|
.map_err(h2_error)?;
|
|
assert_eq!(reason, h2::Reason::CANCEL);
|
|
|
|
let (_, mut final_reply) = next_request(&mut requests).await?;
|
|
let mut final_body = final_reply
|
|
.send_response(response(), false)
|
|
.map_err(h2_error)?;
|
|
final_body
|
|
.send_data(bytes::Bytes::from_static(b"still-open"), true)
|
|
.map_err(h2_error)?;
|
|
task::wait(driver).await?;
|
|
Ok(())
|
|
})
|
|
.await?;
|
|
|
|
let client = http::client(
|
|
&format!("http://localhost:{port}"),
|
|
tls::system(),
|
|
limits(1, 2),
|
|
pool_limits(),
|
|
)
|
|
.await?;
|
|
let first = http::open(&client, request("/cancel")).await?;
|
|
assert_eq!(first.body.pull().await.unwrap()?.values(), b"open");
|
|
assert_eq!(
|
|
http::request(&client, request("/neighbor"))
|
|
.await?
|
|
.body
|
|
.values(),
|
|
b"neighbor"
|
|
);
|
|
first.body.close();
|
|
assert_eq!(
|
|
http::request(&client, request("/after-reset"))
|
|
.await?
|
|
.body
|
|
.values(),
|
|
b"still-open"
|
|
);
|
|
http::close_client(&client).await?;
|
|
task::wait(peer).await?;
|
|
Ok(())
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn stream_limit_blocks_without_expiring_an_active_body_then_reaps_at_eof() {
|
|
run(async {
|
|
let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
|
|
.await
|
|
.map_err(h2_error)?;
|
|
let port = i64::from(listener.local_addr().map_err(h2_error)?.port());
|
|
let finish_first = Channel::new(1)?;
|
|
let finish = finish_first.clone();
|
|
let first_connection_closed = Channel::new(1)?;
|
|
let closed = first_connection_closed.clone();
|
|
let peer = task::run(async move {
|
|
let (first_socket, _) = listener.accept().await.map_err(h2_error)?;
|
|
let (mut first_requests, _shutdown, first_driver) = start_peer(first_socket).await?;
|
|
let (_, mut first_reply) = next_request(&mut first_requests).await?;
|
|
let mut first_body = first_reply
|
|
.send_response(response(), false)
|
|
.map_err(h2_error)?;
|
|
first_body
|
|
.send_data(bytes::Bytes::from_static(b"open"), false)
|
|
.map_err(h2_error)?;
|
|
finish.receive().await?;
|
|
assert!(matches!(
|
|
first_requests.try_recv(),
|
|
Err(mpsc::error::TryRecvError::Empty)
|
|
));
|
|
first_body
|
|
.send_data(bytes::Bytes::from_static(b"done"), true)
|
|
.map_err(h2_error)?;
|
|
task::wait(first_driver).await?;
|
|
closed.send(()).await?;
|
|
|
|
let (second_socket, _) = listener.accept().await.map_err(h2_error)?;
|
|
let (mut second_requests, _shutdown, second_driver) = start_peer(second_socket).await?;
|
|
let (_, mut second_reply) = next_request(&mut second_requests).await?;
|
|
let mut second_body = second_reply
|
|
.send_response(response(), false)
|
|
.map_err(h2_error)?;
|
|
second_body
|
|
.send_data(bytes::Bytes::from_static(b"new"), true)
|
|
.map_err(h2_error)?;
|
|
task::wait(second_driver).await?;
|
|
Ok(())
|
|
})
|
|
.await?;
|
|
|
|
let mut budget = limits(1, 1);
|
|
budget.timeout_ms = 60;
|
|
let mut pool = pool_limits();
|
|
pool.idle_ms = 30;
|
|
let client = http::client(
|
|
&format!("http://localhost:{port}"),
|
|
tls::system(),
|
|
budget,
|
|
pool,
|
|
)
|
|
.await?;
|
|
let first = http::open(&client, request("/active")).await?;
|
|
assert_eq!(first.body.pull().await.unwrap()?.values(), b"open");
|
|
task::sleep(80).await?;
|
|
assert!(
|
|
http::request(&client, request("/blocked"))
|
|
.await
|
|
.unwrap_err()
|
|
.contains("timed out")
|
|
);
|
|
finish_first.send(()).await?;
|
|
assert_eq!(first.body.pull().await.unwrap()?.values(), b"done");
|
|
assert!(first.body.pull().await.is_none());
|
|
first_connection_closed.receive().await?;
|
|
assert_eq!(
|
|
http::request(&client, request("/after-idle"))
|
|
.await?
|
|
.body
|
|
.values(),
|
|
b"new"
|
|
);
|
|
http::close_client(&client).await?;
|
|
task::wait(peer).await?;
|
|
Ok(())
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn goaway_replaces_the_connection_within_the_physical_budget() {
|
|
run(async {
|
|
let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
|
|
.await
|
|
.map_err(h2_error)?;
|
|
let port = i64::from(listener.local_addr().map_err(h2_error)?.port());
|
|
let goaway = Channel::new(1)?;
|
|
let goaway_sent = goaway.clone();
|
|
let finish_old_stream = Channel::new(1)?;
|
|
let finish = finish_old_stream.clone();
|
|
let peer = task::run(async move {
|
|
let (first_socket, _) = listener.accept().await.map_err(h2_error)?;
|
|
let (mut first_requests, shutdown, first_driver) = start_peer(first_socket).await?;
|
|
let (_, mut first_reply) = next_request(&mut first_requests).await?;
|
|
let mut first_body = first_reply
|
|
.send_response(response(), false)
|
|
.map_err(h2_error)?;
|
|
first_body
|
|
.send_data(bytes::Bytes::from_static(b"open"), false)
|
|
.map_err(h2_error)?;
|
|
let (acknowledge, acknowledged) = oneshot::channel();
|
|
shutdown
|
|
.send(acknowledge)
|
|
.map_err(|_| "HTTP/2 peer driver closed before GOAWAY".to_owned())?;
|
|
acknowledged
|
|
.await
|
|
.map_err(|_| "HTTP/2 peer did not begin GOAWAY".to_owned())?;
|
|
task::sleep(100).await?;
|
|
goaway_sent.send(()).await?;
|
|
finish.receive().await?;
|
|
assert!(matches!(
|
|
first_requests.try_recv(),
|
|
Err(mpsc::error::TryRecvError::Empty)
|
|
));
|
|
first_body
|
|
.send_data(bytes::Bytes::from_static(b"done"), true)
|
|
.map_err(h2_error)?;
|
|
task::wait(first_driver).await?;
|
|
|
|
let (second_socket, _) = listener.accept().await.map_err(h2_error)?;
|
|
let (mut second_requests, _shutdown, second_driver) = start_peer(second_socket).await?;
|
|
let (_, mut second_reply) = next_request(&mut second_requests).await?;
|
|
let mut second_body = second_reply
|
|
.send_response(response(), false)
|
|
.map_err(h2_error)?;
|
|
second_body
|
|
.send_data(bytes::Bytes::from_static(b"second"), true)
|
|
.map_err(h2_error)?;
|
|
task::wait(second_driver).await?;
|
|
Ok(())
|
|
})
|
|
.await?;
|
|
|
|
let mut budget = limits(1, 2);
|
|
budget.timeout_ms = 80;
|
|
let client = http::client(
|
|
&format!("http://localhost:{port}"),
|
|
tls::system(),
|
|
budget,
|
|
pool_limits(),
|
|
)
|
|
.await?;
|
|
let first = http::open(&client, request("/before-goaway")).await?;
|
|
assert_eq!(first.body.pull().await.unwrap()?.values(), b"open");
|
|
goaway.receive().await?;
|
|
let blocked = http::request(&client, request("/must-wait"))
|
|
.await
|
|
.unwrap_err();
|
|
assert!(blocked.contains("timed out"), "{blocked}");
|
|
finish_old_stream.send(()).await?;
|
|
assert_eq!(first.body.pull().await.unwrap()?.values(), b"done");
|
|
assert!(first.body.pull().await.is_none());
|
|
assert_eq!(
|
|
http::request(&client, request("/after-goaway"))
|
|
.await?
|
|
.body
|
|
.values(),
|
|
b"second"
|
|
);
|
|
http::close_client(&client).await?;
|
|
task::wait(peer).await?;
|
|
Ok(())
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn https_requires_h2_alpn_without_protocol_fallback() {
|
|
run(async {
|
|
let listener = net::listen("127.0.0.1", 0).await?;
|
|
let port = net::port(&listener)?;
|
|
let mut http1 = limits(1, 1);
|
|
http1.http2 = None;
|
|
let server_listener = listener.clone();
|
|
let server = task::run(async move {
|
|
http::serve_tls(
|
|
|_| async {
|
|
Ok(http::Response {
|
|
status: 200,
|
|
headers: vec![],
|
|
body: bytes(b"unexpected"),
|
|
})
|
|
},
|
|
server_listener,
|
|
http1,
|
|
identity(),
|
|
)
|
|
.await
|
|
})
|
|
.await?;
|
|
let client = http::client(
|
|
&format!("https://localhost:{port}"),
|
|
trust(),
|
|
limits(1, 1),
|
|
pool_limits(),
|
|
)
|
|
.await?;
|
|
let error = http::request(&client, request("/alpn")).await.unwrap_err();
|
|
assert!(error.contains("NoApplicationProtocol"), "{error}");
|
|
http::close_client(&client).await?;
|
|
task::stop(server).await?;
|
|
listener.close()?;
|
|
Ok(())
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn invalid_http2_budgets_and_outgoing_headers_fail_before_connecting() {
|
|
run(async {
|
|
let base = limits(1, 2);
|
|
let mut cases = Vec::new();
|
|
for streams in [0, 65537] {
|
|
let mut budget = base;
|
|
budget.http2.as_mut().unwrap().streams = streams;
|
|
cases.push((budget, "connections * streams"));
|
|
}
|
|
let mut multiplied = base;
|
|
multiplied.connections = 32769;
|
|
cases.push((multiplied, "connections * streams"));
|
|
for window in [0, i64::from(i32::MAX) + 1] {
|
|
let mut budget = base;
|
|
budget.http2.as_mut().unwrap().stream_window_bytes = window;
|
|
cases.push((budget, "stream_window_bytes"));
|
|
}
|
|
for window in [65534, i64::from(i32::MAX) + 1] {
|
|
let mut budget = base;
|
|
budget.http2.as_mut().unwrap().connection_window_bytes = window;
|
|
cases.push((budget, "connection_window_bytes"));
|
|
}
|
|
for (budget, expected) in cases {
|
|
let failure = http::client("http://127.0.0.1:1", tls::system(), budget, pool_limits())
|
|
.await
|
|
.unwrap_err();
|
|
assert!(failure.contains(expected), "{failure}");
|
|
}
|
|
|
|
let client = http::client("http://127.0.0.1:1", tls::system(), base, pool_limits()).await?;
|
|
for (name, value) in [
|
|
("Connection", "close"),
|
|
("Keep-Alive", "timeout=5"),
|
|
("TE", "gzip"),
|
|
] {
|
|
let mut outgoing = request("/");
|
|
outgoing.headers.push(http::Header {
|
|
name: name.into(),
|
|
value: bytes(value.as_bytes()),
|
|
});
|
|
let failure = http::request(&client, outgoing).await.unwrap_err();
|
|
assert!(
|
|
failure.contains("HTTP/2 forbids connection-specific header"),
|
|
"{failure}"
|
|
);
|
|
}
|
|
http::close_client(&client).await?;
|
|
Ok(())
|
|
});
|
|
}
|