use std::{
borrow::Cow,
pin::Pin,
sync::Arc,
task::{Context, Poll},
};
use bytes::Bytes;
use dhttp_identity::identity::{self as authority, SignError};
use futures::{Sink, Stream, future::BoxFuture};
use rustls::{pki_types::CertificateDer, sign::CertifiedKey};
use crate::{error::Code, quic, quic::ResetStream, varint::VarInt};
pub fn convert_varint(varint: dquic::prelude::VarInt) -> VarInt {
VarInt::from_u64(varint.into_u64()).expect("dquic VarInt is within valid range")
}
pub fn convert_connection_error(error: dquic::prelude::Error) -> quic::ConnectionError {
match error {
dquic::prelude::Error::Quic(quic_error) => {
quic::ConnectionError::from(quic::TransportError {
kind: convert_varint(quic_error.kind().into()),
frame_type: match quic_error.frame_type() {
dquic::qbase::error::ErrorFrameType::V1(frame_type) => {
convert_varint(frame_type.into())
}
dquic::qbase::error::ErrorFrameType::Ext(varint) => convert_varint(varint),
},
reason: quic_error.reason().to_owned().into(),
})
}
dquic::prelude::Error::App(app_error) => {
quic::ConnectionError::from(quic::ApplicationError {
code: Code::new(
VarInt::try_from(app_error.error_code())
.expect("QUIC application error code fits in VarInt range"),
),
reason: app_error.reason().to_owned().into(),
})
}
}
}
pub fn convert_stream_error(error: dquic::prelude::StreamError) -> quic::StreamError {
match error {
dquic::prelude::StreamError::Connection(error) => {
quic::StreamError::from(convert_connection_error(error))
}
dquic::prelude::StreamError::Reset(reset_stream_error) => quic::StreamError::Reset {
code: VarInt::try_from(reset_stream_error.error_code())
.expect("QUIC reset error code fits in VarInt range"),
},
dquic::prelude::StreamError::EosSent => {
panic!("h3x write data after shutdown")
}
}
}
pin_project_lite::pin_project! {
pub struct StreamReader {
pub(super) stream_id: VarInt,
#[pin]
pub(super) reader: dquic::prelude::StreamReader,
}
}
impl quic::GetStreamId for StreamReader {
fn poll_stream_id(
self: Pin<&mut Self>,
_: &mut Context,
) -> Poll<Result<VarInt, quic::StreamError>> {
Poll::Ready(Ok(self.stream_id))
}
}
impl Stream for StreamReader {
type Item = Result<Bytes, quic::StreamError>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
self.project()
.reader
.poll_next(cx)
.map_ok(|bytes| bytes)
.map_err(convert_stream_error)
}
fn size_hint(&self) -> (usize, Option<usize>) {
self.reader.size_hint()
}
}
impl quic::StopStream for StreamReader {
fn poll_stop(
self: Pin<&mut Self>,
_: &mut Context,
code: VarInt,
) -> Poll<Result<(), quic::StreamError>> {
dquic::prelude::StopSending::stop(&mut self.get_mut().reader, code.into_inner());
Poll::Ready(Ok(()))
}
}
pin_project_lite::pin_project! {
pub struct StreamWriter {
pub(super) stream_id: VarInt,
#[pin]
pub(super) writer: dquic::prelude::StreamWriter,
}
}
impl quic::GetStreamId for StreamWriter {
fn poll_stream_id(
self: Pin<&mut Self>,
_: &mut Context,
) -> Poll<Result<VarInt, quic::StreamError>> {
Poll::Ready(Ok(self.stream_id))
}
}
impl Sink<Bytes> for StreamWriter {
type Error = quic::StreamError;
fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.project()
.writer
.poll_ready(cx)
.map_err(convert_stream_error)
}
fn start_send(self: Pin<&mut Self>, item: Bytes) -> Result<(), Self::Error> {
self.project()
.writer
.start_send(item)
.map_err(convert_stream_error)
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
_ = cx;
Poll::Ready(Ok(()))
}
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.project()
.writer
.poll_close(cx)
.map_err(convert_stream_error)
}
}
impl ResetStream for StreamWriter {
fn poll_reset(
self: Pin<&mut Self>,
_: &mut Context,
code: VarInt,
) -> Poll<Result<(), quic::StreamError>> {
dquic::prelude::CancelStream::cancel(&mut self.get_mut().writer, code.into_inner());
Poll::Ready(Ok(()))
}
}
impl quic::ManageStream for dquic::prelude::Connection {
type StreamWriter = StreamWriter;
type StreamReader = StreamReader;
async fn open_bi(
&self,
) -> Result<(Self::StreamReader, Self::StreamWriter), quic::ConnectionError> {
let stream = self
.open_bi_stream()
.await
.map_err(convert_connection_error)?;
let (stream_id, (reader, writer)) = stream.ok_or_else(|| {
quic::ConnectionError::from(quic::TransportError {
kind: VarInt::from_u32(0x04), frame_type: VarInt::from_u32(0),
reason: "stream ID space exhausted".into(),
})
})?;
let stream_id = convert_varint(stream_id.into());
let reader = StreamReader { stream_id, reader };
let writer = StreamWriter { stream_id, writer };
Ok((reader, writer))
}
async fn open_uni(&self) -> Result<Self::StreamWriter, quic::ConnectionError> {
let stream = self
.open_uni_stream()
.await
.map_err(convert_connection_error)?;
let (stream_id, writer) = stream.ok_or_else(|| {
quic::ConnectionError::from(quic::TransportError {
kind: VarInt::from_u32(0x04), frame_type: VarInt::from_u32(0),
reason: "stream ID space exhausted".into(),
})
})?;
let stream_id = convert_varint(stream_id.into());
Ok(StreamWriter { stream_id, writer })
}
async fn accept_bi(
&self,
) -> Result<(Self::StreamReader, Self::StreamWriter), quic::ConnectionError> {
let (stream_id, (reader, writer)) = self
.accept_bi_stream()
.await
.map_err(convert_connection_error)?;
let stream_id = convert_varint(stream_id.into());
let reader = StreamReader { stream_id, reader };
let writer = StreamWriter { stream_id, writer };
Ok((reader, writer))
}
async fn accept_uni(&self) -> Result<Self::StreamReader, quic::ConnectionError> {
let (stream_id, reader) = self
.accept_uni_stream()
.await
.map_err(convert_connection_error)?;
let stream_id = convert_varint(stream_id.into());
Ok(StreamReader { stream_id, reader })
}
}
#[derive(Debug)]
pub struct DquicLocalAuthority {
name: Arc<str>,
certified_key: Arc<CertifiedKey>,
}
impl authority::LocalAuthority for DquicLocalAuthority {
fn name(&self) -> &str {
&self.name
}
fn cert_chain(&self) -> &[CertificateDer<'static>] {
self.certified_key.cert.as_slice()
}
fn sign(&self, data: &[u8]) -> BoxFuture<'_, Result<Vec<u8>, SignError>> {
let result = authority::sign_with_key(self.certified_key.key.as_ref(), data);
Box::pin(std::future::ready(result))
}
}
#[derive(Debug)]
pub struct DquicRemoteAuthority {
name: Arc<str>,
cert_chain: Arc<[CertificateDer<'static>]>,
}
impl authority::RemoteAuthority for DquicRemoteAuthority {
fn name(&self) -> &str {
&self.name
}
fn cert_chain(&self) -> &[CertificateDer<'static>] {
&self.cert_chain
}
}
impl quic::WithLocalAuthority for dquic::prelude::Connection {
type LocalAuthority = DquicLocalAuthority;
async fn local_authority(&self) -> Result<Option<DquicLocalAuthority>, quic::ConnectionError> {
let authority = dquic::prelude::Connection::local_authority(self)
.await
.map_err(convert_connection_error)?;
Ok(authority.map(|authority| {
let name = AsRef::<Arc<str>>::as_ref(&authority).clone();
let certified_key = AsRef::<Arc<CertifiedKey>>::as_ref(&authority).clone();
DquicLocalAuthority {
name,
certified_key,
}
}))
}
}
impl quic::WithRemoteAuthority for dquic::prelude::Connection {
type RemoteAuthority = DquicRemoteAuthority;
async fn remote_authority(
&self,
) -> Result<Option<DquicRemoteAuthority>, quic::ConnectionError> {
let authority = dquic::prelude::Connection::remote_authority(self)
.await
.map_err(convert_connection_error)?;
Ok(authority.map(|authority| {
let name = AsRef::<Arc<str>>::as_ref(&authority).clone();
let cert_chain = AsRef::<Arc<[CertificateDer]>>::as_ref(&authority).clone();
DquicRemoteAuthority { name, cert_chain }
}))
}
}
impl quic::Lifecycle for dquic::prelude::Connection {
fn close(&self, code: Code, reason: Cow<'static, str>) {
_ = dquic::prelude::Connection::close(self, reason, code.into_inner().into_inner());
}
fn check(&self) -> Result<(), quic::ConnectionError> {
self.validate().map_err(convert_connection_error)
}
async fn closed(&self) -> quic::ConnectionError {
convert_connection_error(dquic::prelude::Connection::terminated(self).await)
}
}
impl quic::Listen for dquic::prelude::QuicListeners {
type Connection = dquic::prelude::Connection;
type Error = dquic::prelude::ListenersShutdown;
async fn accept(&mut self) -> Result<Arc<Self::Connection>, Self::Error> {
let (connection, ..) = dquic::prelude::QuicListeners::accept(self).await?;
Ok(connection)
}
async fn shutdown(&self) -> Result<(), Self::Error> {
dquic::prelude::QuicListeners::shutdown(self);
Ok(())
}
}
impl quic::Listen for &dquic::prelude::QuicListeners {
type Connection = dquic::prelude::Connection;
type Error = dquic::prelude::ListenersShutdown;
async fn accept(&mut self) -> Result<Arc<Self::Connection>, Self::Error> {
let (connection, ..) = (*self).accept().await?;
Ok(connection)
}
async fn shutdown(&self) -> Result<(), Self::Error> {
(*self).shutdown();
Ok(())
}
}
impl quic::Listen for Arc<dquic::prelude::QuicListeners> {
type Connection = dquic::prelude::Connection;
type Error = dquic::prelude::ListenersShutdown;
async fn accept(&mut self) -> Result<Arc<Self::Connection>, Self::Error> {
let (connection, ..) = self.as_ref().accept().await?;
Ok(connection)
}
async fn shutdown(&self) -> Result<(), Self::Error> {
self.as_ref().shutdown();
Ok(())
}
}
impl quic::Connect for Arc<dquic::prelude::QuicClient> {
type Connection = dquic::prelude::Connection;
type Error = dquic::prelude::ConnectServerError;
async fn connect(
&self,
server: &http::uri::Authority,
) -> Result<Arc<Self::Connection>, Self::Error> {
let name = if let Some(port) = server.port_u16() {
format!("{}:{port}", server.host())
} else {
server.host().to_owned()
};
let connection = dquic::prelude::QuicClient::connect(self, &name).await?;
Ok(connection)
}
}
#[cfg(test)]
mod tests {
use std::{
borrow::Cow,
fmt,
str::FromStr,
sync::{Arc, Mutex},
time::Duration,
};
use dquic::{
prelude::{
Connection, IO,
handy::{ToCertificate, ToPrivateKey},
},
qbase::{
error::{AppError, Error as DquicError, ErrorFrameType, ErrorKind, QuicError},
frame::{FrameType, ResetStreamError},
varint::VarInt as DquicVarInt,
},
qrecovery::streams::error::StreamError as DquicStreamError,
};
use futures::{FutureExt, SinkExt, StreamExt, stream};
use http::uri::Authority;
use tokio::time;
use super::*;
use crate::{
dquic::resolver::{Record, Resolve, ResolveFuture},
quic::{GetStreamIdExt, Lifecycle, ResetStreamExt, StopStreamExt},
};
const SERVER_CERT: &[u8] = include_bytes!("../../tests/keychain/localhost/server.cert");
const SERVER_KEY: &[u8] = include_bytes!("../../tests/keychain/localhost/server.key");
fn make_identity() -> crate::dquic::identity::Identity {
crate::dquic::identity::Identity::new(
"localhost".parse().expect("valid identity name"),
SERVER_CERT.to_certificate(),
SERVER_KEY.to_private_key(),
)
}
const TEST_TIMEOUT: Duration = Duration::from_secs(10);
async fn with_timeout<T>(future: impl Future<Output = T>) -> T {
time::timeout(TEST_TIMEOUT, future)
.await
.expect("test operation timed out")
}
async fn make_server_endpoint() -> (crate::dquic::QuicEndpoint, Authority) {
let identity = Arc::new(make_identity());
let network = crate::dquic::Network::builder().build();
let endpoint = crate::dquic::QuicEndpoint::builder()
.network(network.clone())
.identity(identity)
.bind(Arc::new(vec![
crate::dquic::binds::BindPattern::from_str("127.0.0.1:0")
.expect("valid bind pattern"),
]))
.build()
.await;
let bind_iface = network
.quic()
.interfaces()
.into_iter()
.next()
.expect("server should bind an interface");
let port = bind_iface
.borrow()
.bound_addr()
.expect("server interface should have a bound address")
.port();
let authority =
Authority::from_maybe_shared(format!("localhost:{port}")).expect("valid authority");
(endpoint, authority)
}
fn make_raw_listeners() -> Arc<dquic::prelude::QuicListeners> {
dquic::prelude::QuicListeners::builder()
.with_router(Arc::new(
dquic::qinterface::component::route::QuicRouter::default(),
))
.with_local_endpoints(Arc::new(
dquic::qinterface::component::local_endpoint::LocalEndpoints::new(),
))
.without_client_cert_verifier()
.listen(1)
.expect("raw listeners start")
}
struct ConnectedPair {
_client_endpoint: crate::dquic::QuicEndpoint,
_server_endpoint: crate::dquic::QuicEndpoint,
client: Arc<Connection>,
server: Arc<Connection>,
}
async fn connected_pair() -> ConnectedPair {
let client_endpoint = crate::dquic::QuicEndpoint::builder().build().await;
let (server_endpoint, authority) = make_server_endpoint().await;
let (server_connection, client_connection) = with_timeout(async {
tokio::join!(
server_endpoint.accept(),
quic::Connect::connect(&client_endpoint, &authority)
)
})
.await;
ConnectedPair {
_client_endpoint: client_endpoint,
_server_endpoint: server_endpoint,
client: client_connection.expect("client should connect"),
server: server_connection.expect("server should accept"),
}
}
fn close_pair(pair: &ConnectedPair) {
Lifecycle::close(pair.client.as_ref(), Code::H3_NO_ERROR, Cow::Borrowed(""));
Lifecycle::close(pair.server.as_ref(), Code::H3_NO_ERROR, Cow::Borrowed(""));
}
#[derive(Debug, Default)]
struct RecordingResolver {
lookups: Mutex<Vec<(String, String, Option<crate::dquic::qresolve::Family>)>>,
}
impl RecordingResolver {
fn lookups(&self) -> Vec<(String, String, Option<crate::dquic::qresolve::Family>)> {
self.lookups.lock().unwrap().clone()
}
}
impl fmt::Display for RecordingResolver {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("recording resolver")
}
}
impl Resolve for RecordingResolver {
fn lookup<'l>(
&'l self,
hostname: &'l str,
servname: &'l str,
family: Option<crate::dquic::qresolve::Family>,
) -> ResolveFuture<'l> {
self.lookups
.lock()
.unwrap()
.push((hostname.to_owned(), servname.to_owned(), family));
async { Ok(stream::empty::<Record>().boxed()) }.boxed()
}
}
fn assert_transport_error(
error: quic::ConnectionError,
expected_kind: VarInt,
expected_frame_type: VarInt,
expected_reason: &str,
) {
let quic::ConnectionError::Transport { source } = error else {
panic!("expected transport error");
};
assert_eq!(source.kind, expected_kind);
assert_eq!(source.frame_type, expected_frame_type);
assert_eq!(source.reason, Cow::Borrowed(expected_reason));
}
#[test]
fn convert_varint_preserves_dquic_value() {
let value = DquicVarInt::from_u64(0x1234_5678).expect("dquic varint");
assert_eq!(convert_varint(value), VarInt::from_u32(0x1234_5678));
}
#[test]
fn convert_connection_error_preserves_transport_v1_frame_type() {
let error = DquicError::Quic(QuicError::new(
ErrorKind::StreamLimit,
ErrorFrameType::V1(FrameType::StopSending),
"stream limit",
));
assert_transport_error(
convert_connection_error(error),
VarInt::from_u32(0x04),
VarInt::from_u32(0x05),
"stream limit",
);
}
#[test]
fn convert_connection_error_preserves_transport_extension_frame_type() {
let extension_frame = DquicVarInt::from_u32(0x21);
let error = DquicError::Quic(QuicError::new(
ErrorKind::ProtocolViolation,
ErrorFrameType::Ext(extension_frame),
"extension frame",
));
assert_transport_error(
convert_connection_error(error),
VarInt::from_u32(0x0a),
VarInt::from_u32(0x21),
"extension frame",
);
}
#[test]
fn convert_connection_error_preserves_application_error() {
let error = DquicError::App(AppError::new(
DquicVarInt::from_u32(Code::H3_REQUEST_CANCELLED.into_inner().into_inner() as u32),
"application close",
));
let quic::ConnectionError::Application { source } = convert_connection_error(error) else {
panic!("expected application error");
};
assert_eq!(source.code, Code::H3_REQUEST_CANCELLED);
assert_eq!(source.reason, Cow::Borrowed("application close"));
}
#[test]
fn convert_stream_error_preserves_connection_and_reset_errors() {
let connection = DquicStreamError::Connection(DquicError::Quic(
QuicError::with_default_fty(ErrorKind::Internal, "connection failed"),
));
let quic::StreamError::Connection { source } = convert_stream_error(connection) else {
panic!("expected connection stream error");
};
assert_transport_error(
source,
VarInt::from_u32(0x01),
VarInt::from_u32(0x00),
"connection failed",
);
let reset = DquicStreamError::Reset(ResetStreamError::new(
DquicVarInt::from_u32(0x100),
DquicVarInt::from_u32(0),
));
let quic::StreamError::Reset { code } = convert_stream_error(reset) else {
panic!("expected reset stream error");
};
assert_eq!(code, VarInt::from_u32(0x100));
}
#[test]
#[should_panic(expected = "h3x write data after shutdown")]
fn convert_stream_error_rejects_eos_sent() {
_ = convert_stream_error(DquicStreamError::EosSent);
}
#[tokio::test]
async fn connection_agents_are_exposed_from_established_dquic_connection() {
let pair = connected_pair().await;
let client = &pair.client;
let server = &pair.server;
let client_local = quic::WithLocalAuthority::local_authority(client.as_ref())
.await
.expect("client local authority lookup");
assert!(
client_local.is_none(),
"anonymous client has no local authority"
);
let client_remote = quic::WithRemoteAuthority::remote_authority(client.as_ref())
.await
.expect("client remote authority lookup")
.expect("client should observe server identity");
assert_eq!(
authority::RemoteAuthority::name(&client_remote),
"localhost"
);
assert_eq!(
authority::RemoteAuthority::cert_chain(&client_remote),
make_identity().cert_chain()
);
let server_local = quic::WithLocalAuthority::local_authority(server.as_ref())
.await
.expect("server local authority lookup")
.expect("server should have local identity");
assert_eq!(authority::LocalAuthority::name(&server_local), "localhost");
assert_eq!(
authority::LocalAuthority::cert_chain(&server_local),
make_identity().cert_chain()
);
let server_remote = quic::WithRemoteAuthority::remote_authority(server.as_ref())
.await
.expect("server remote authority lookup");
assert!(
server_remote.is_none(),
"server should not observe an anonymous client identity"
);
}
#[tokio::test]
async fn lifecycle_close_check_and_closed_report_terminal_error() {
let pair = connected_pair().await;
let client = &pair.client;
let server = &pair.server;
Lifecycle::check(client.as_ref()).expect("client initially live");
Lifecycle::check(server.as_ref()).expect("server initially live");
Lifecycle::close(
client.as_ref(),
Code::H3_REQUEST_CANCELLED,
Cow::Borrowed("done"),
);
let error = with_timeout(Lifecycle::closed(server.as_ref())).await;
assert!(
error.is_application() || error.is_transport(),
"closed should report a terminal connection error"
);
let checked = Lifecycle::check(server.as_ref()).expect_err("terminal error is sticky");
assert_eq!(checked.to_string(), error.to_string());
let repeated = Lifecycle::closed(server.as_ref())
.now_or_never()
.expect("terminal closed should resolve immediately");
assert_eq!(repeated.to_string(), error.to_string());
}
#[tokio::test]
async fn listen_impls_for_references_and_arcs_shutdown_accept_queue() {
let listeners = make_raw_listeners();
quic::Listen::shutdown(listeners.as_ref())
.await
.expect("direct listener shutdown");
let mut by_ref = listeners.as_ref();
quic::Listen::shutdown(&by_ref)
.await
.expect("reference listener shutdown");
assert!(quic::Listen::accept(&mut by_ref).await.is_err());
let mut by_arc = listeners.clone();
quic::Listen::shutdown(&by_arc)
.await
.expect("arc listener shutdown");
assert!(quic::Listen::accept(&mut by_arc).await.is_err());
}
#[tokio::test]
async fn isolated_raw_listener_configs_allow_concurrent_listener_instances() {
let first = make_raw_listeners();
let second = make_raw_listeners();
quic::Listen::shutdown(first.as_ref())
.await
.expect("first isolated listener shutdown");
quic::Listen::shutdown(second.as_ref())
.await
.expect("second isolated listener shutdown");
}
#[tokio::test]
async fn connect_impl_formats_authority_with_and_without_port() {
let resolver = Arc::new(RecordingResolver::default());
let client = Arc::new(
dquic::prelude::QuicClient::builder()
.with_resolver(resolver.clone())
.without_verifier()
.without_cert()
.build(),
);
let with_port = "example.test:8443"
.parse::<Authority>()
.expect("authority with port parses");
let without_port = "example.test"
.parse::<Authority>()
.expect("authority without port parses");
let first = quic::Connect::connect(&client, &with_port)
.await
.expect("connect with port");
let second = quic::Connect::connect(&client, &without_port)
.await
.expect("connect without port");
assert_eq!(
resolver.lookups(),
vec![
("example.test:8443".to_owned(), "443".to_owned(), None),
("example.test".to_owned(), "443".to_owned(), None),
]
);
assert_eq!(format!("{}", resolver.as_ref()), "recording resolver");
Lifecycle::close(first.as_ref(), Code::H3_NO_ERROR, Cow::Borrowed(""));
Lifecycle::close(second.as_ref(), Code::H3_NO_ERROR, Cow::Borrowed(""));
}
#[tokio::test]
async fn dquic_local_authority_exposes_identity_and_signing() {
let identity = make_identity();
let certified_key =
crate::dquic::identity::build_certified_key(&identity).expect("test key should load");
let local = DquicLocalAuthority {
name: Arc::from(identity.name.as_str()),
certified_key,
};
assert_eq!(authority::LocalAuthority::name(&local), "localhost");
assert_eq!(
authority::LocalAuthority::cert_chain(&local),
identity.cert_chain()
);
assert!(format!("{local:?}").contains("DquicLocalAuthority"));
let signature = authority::LocalAuthority::sign(&local, b"payload")
.await
.expect("signature");
assert!(
authority::LocalAuthority::verify(&local, b"payload", &signature)
.await
.expect("verification should run")
);
assert!(
!authority::LocalAuthority::verify(&local, b"wrong payload", &signature)
.await
.expect("verification should run")
);
}
#[tokio::test]
async fn dquic_remote_authority_exposes_identity_and_verification() {
let identity = make_identity();
let certified_key =
crate::dquic::identity::build_certified_key(&identity).expect("test key should load");
let local = DquicLocalAuthority {
name: Arc::from(identity.name.as_str()),
certified_key,
};
let remote = DquicRemoteAuthority {
name: Arc::from("peer.localhost"),
cert_chain: Arc::from(identity.cert_chain()),
};
assert_eq!(authority::RemoteAuthority::name(&remote), "peer.localhost");
assert_eq!(
authority::RemoteAuthority::cert_chain(&remote),
identity.cert_chain()
);
assert_eq!(
authority::RemoteAuthority::public_key(&remote).as_ref(),
authority::LocalAuthority::public_key(&local).as_ref()
);
assert!(format!("{remote:?}").contains("DquicRemoteAuthority"));
let signature = authority::LocalAuthority::sign(&local, b"payload")
.await
.expect("signature");
assert!(
authority::RemoteAuthority::verify(&remote, b"payload", &signature)
.await
.expect("verification should run")
);
assert!(
!authority::RemoteAuthority::verify(&remote, b"wrong payload", &signature)
.await
.expect("verification should run")
);
}
#[tokio::test]
async fn bidirectional_stream_wrappers_expose_ids_and_transfer_bytes() {
let pair = connected_pair().await;
let (mut client_reader, mut client_writer) =
with_timeout(quic::ManageStream::open_bi(pair.client.as_ref()))
.await
.expect("client opens stream");
let client_reader_id = client_reader.stream_id().await.expect("client reader id");
let client_writer_id = client_writer.stream_id().await.expect("client writer id");
assert_eq!(client_reader_id, client_writer_id);
client_writer
.send(Bytes::from_static(b"client payload"))
.await
.expect("client sends request bytes");
let (mut server_reader, mut server_writer) =
with_timeout(quic::ManageStream::accept_bi(pair.server.as_ref()))
.await
.expect("server accepts stream");
let server_reader_id = server_reader.stream_id().await.expect("server reader id");
let server_writer_id = server_writer.stream_id().await.expect("server writer id");
assert_eq!(server_reader_id, server_writer_id);
assert_eq!(client_reader_id, server_reader_id);
assert_eq!(server_reader.size_hint(), (0, None));
let received = with_timeout(server_reader.next())
.await
.expect("server receives a chunk")
.expect("server chunk is ok");
assert_eq!(received, Bytes::from_static(b"client payload"));
server_writer
.send(Bytes::from_static(b"server payload"))
.await
.expect("server sends response bytes");
let received = with_timeout(client_reader.next())
.await
.expect("client receives a chunk")
.expect("client chunk is ok");
assert_eq!(received, Bytes::from_static(b"server payload"));
with_timeout(client_writer.close())
.await
.expect("client writer closes");
with_timeout(server_writer.close())
.await
.expect("server writer closes");
close_pair(&pair);
}
#[tokio::test]
async fn unidirectional_stream_wrappers_expose_ids_and_transfer_bytes() {
let pair = connected_pair().await;
let mut client_writer = with_timeout(quic::ManageStream::open_uni(pair.client.as_ref()))
.await
.expect("client opens unidirectional stream");
let client_stream_id = client_writer.stream_id().await.expect("client writer id");
client_writer
.send(Bytes::from_static(b"one-way payload"))
.await
.expect("client sends unidirectional bytes");
let mut server_reader = with_timeout(quic::ManageStream::accept_uni(pair.server.as_ref()))
.await
.expect("server accepts unidirectional stream");
let server_stream_id = server_reader.stream_id().await.expect("server reader id");
assert_eq!(client_stream_id, server_stream_id);
let received = with_timeout(server_reader.next())
.await
.expect("server receives unidirectional chunk")
.expect("server unidirectional chunk is ok");
assert_eq!(received, Bytes::from_static(b"one-way payload"));
with_timeout(client_writer.close())
.await
.expect("client unidirectional writer closes");
close_pair(&pair);
}
#[tokio::test]
async fn stream_stop_and_reset_wrappers_complete() {
let pair = connected_pair().await;
let mut client_writer = with_timeout(quic::ManageStream::open_uni(pair.client.as_ref()))
.await
.expect("client opens stream for stop");
client_writer
.send(Bytes::from_static(b"stop payload"))
.await
.expect("client sends bytes before stop");
let mut server_reader = with_timeout(quic::ManageStream::accept_uni(pair.server.as_ref()))
.await
.expect("server accepts stream for stop");
server_reader
.stop(VarInt::from_u32(0x10))
.await
.expect("stop-sending completes on reader wrapper");
client_writer
.reset(VarInt::from_u32(0x11))
.await
.expect("reset completes on writer wrapper");
close_pair(&pair);
}
}