mirror of
https://boringssl.googlesource.com/boringssl
synced 2026-07-21 14:43:51 +00:00
rust: bssl-tls: Carve out tokio support into its own crate
Signed-off-by: Xiangfei Ding <xfding@google.com> Change-Id: I88b143506455afdb6c51d3914eddc0b26a6a6964 Reviewed-on: https://boringssl-review.googlesource.com/c/boringssl/+/94508 Reviewed-by: Adam Langley <agl@google.com> Presubmit-BoringSSL-Verified: boringssl-scoped@luci-project-accounts.iam.gserviceaccount.com <boringssl-scoped@luci-project-accounts.iam.gserviceaccount.com>
This commit is contained in:
Xiangfei Ding
Generated
+22
-87
@@ -49,11 +49,20 @@ dependencies = [
|
||||
"futures",
|
||||
"libc",
|
||||
"once_cell",
|
||||
"tokio",
|
||||
"tracing",
|
||||
"tracing-subscriber",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "bssl-tls-tokio"
|
||||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"bssl-tls",
|
||||
"bssl-x509",
|
||||
"futures",
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "bssl-x509"
|
||||
version = "0.1.0"
|
||||
@@ -425,12 +434,12 @@ checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03"
|
||||
|
||||
[[package]]
|
||||
name = "socket2"
|
||||
version = "0.6.2"
|
||||
version = "0.6.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "86f4aa3ad99f2088c990dfa82d367e19cb29268ed67c574d10d0a4bfe71f07e0"
|
||||
checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"windows-sys 0.60.2",
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -580,16 +589,7 @@ version = "0.52.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d"
|
||||
dependencies = [
|
||||
"windows-targets 0.52.6",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "windows-sys"
|
||||
version = "0.60.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb"
|
||||
dependencies = [
|
||||
"windows-targets 0.53.5",
|
||||
"windows-targets",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -607,31 +607,14 @@ version = "0.52.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973"
|
||||
dependencies = [
|
||||
"windows_aarch64_gnullvm 0.52.6",
|
||||
"windows_aarch64_msvc 0.52.6",
|
||||
"windows_i686_gnu 0.52.6",
|
||||
"windows_i686_gnullvm 0.52.6",
|
||||
"windows_i686_msvc 0.52.6",
|
||||
"windows_x86_64_gnu 0.52.6",
|
||||
"windows_x86_64_gnullvm 0.52.6",
|
||||
"windows_x86_64_msvc 0.52.6",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "windows-targets"
|
||||
version = "0.53.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3"
|
||||
dependencies = [
|
||||
"windows-link",
|
||||
"windows_aarch64_gnullvm 0.53.1",
|
||||
"windows_aarch64_msvc 0.53.1",
|
||||
"windows_i686_gnu 0.53.1",
|
||||
"windows_i686_gnullvm 0.53.1",
|
||||
"windows_i686_msvc 0.53.1",
|
||||
"windows_x86_64_gnu 0.53.1",
|
||||
"windows_x86_64_gnullvm 0.53.1",
|
||||
"windows_x86_64_msvc 0.53.1",
|
||||
"windows_aarch64_gnullvm",
|
||||
"windows_aarch64_msvc",
|
||||
"windows_i686_gnu",
|
||||
"windows_i686_gnullvm",
|
||||
"windows_i686_msvc",
|
||||
"windows_x86_64_gnu",
|
||||
"windows_x86_64_gnullvm",
|
||||
"windows_x86_64_msvc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -640,96 +623,48 @@ version = "0.52.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3"
|
||||
|
||||
[[package]]
|
||||
name = "windows_aarch64_gnullvm"
|
||||
version = "0.53.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53"
|
||||
|
||||
[[package]]
|
||||
name = "windows_aarch64_msvc"
|
||||
version = "0.52.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469"
|
||||
|
||||
[[package]]
|
||||
name = "windows_aarch64_msvc"
|
||||
version = "0.53.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006"
|
||||
|
||||
[[package]]
|
||||
name = "windows_i686_gnu"
|
||||
version = "0.52.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b"
|
||||
|
||||
[[package]]
|
||||
name = "windows_i686_gnu"
|
||||
version = "0.53.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "960e6da069d81e09becb0ca57a65220ddff016ff2d6af6a223cf372a506593a3"
|
||||
|
||||
[[package]]
|
||||
name = "windows_i686_gnullvm"
|
||||
version = "0.52.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66"
|
||||
|
||||
[[package]]
|
||||
name = "windows_i686_gnullvm"
|
||||
version = "0.53.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c"
|
||||
|
||||
[[package]]
|
||||
name = "windows_i686_msvc"
|
||||
version = "0.52.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66"
|
||||
|
||||
[[package]]
|
||||
name = "windows_i686_msvc"
|
||||
version = "0.53.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2"
|
||||
|
||||
[[package]]
|
||||
name = "windows_x86_64_gnu"
|
||||
version = "0.52.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78"
|
||||
|
||||
[[package]]
|
||||
name = "windows_x86_64_gnu"
|
||||
version = "0.53.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499"
|
||||
|
||||
[[package]]
|
||||
name = "windows_x86_64_gnullvm"
|
||||
version = "0.52.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d"
|
||||
|
||||
[[package]]
|
||||
name = "windows_x86_64_gnullvm"
|
||||
version = "0.53.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1"
|
||||
|
||||
[[package]]
|
||||
name = "windows_x86_64_msvc"
|
||||
version = "0.52.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec"
|
||||
|
||||
[[package]]
|
||||
name = "windows_x86_64_msvc"
|
||||
version = "0.53.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650"
|
||||
|
||||
[[package]]
|
||||
name = "zeroize"
|
||||
version = "1.8.2"
|
||||
|
||||
@@ -5,6 +5,7 @@ members = [
|
||||
"bssl-rustls-adapters",
|
||||
"bssl-sys",
|
||||
"bssl-tls",
|
||||
"bssl-tls-tokio",
|
||||
"bssl-x509",
|
||||
]
|
||||
resolver = "3"
|
||||
|
||||
@@ -0,0 +1,15 @@
|
||||
[package]
|
||||
name = "bssl-tls-tokio"
|
||||
version = "0.1.0"
|
||||
edition = "2024"
|
||||
publish = false
|
||||
license = "Apache-2.0"
|
||||
|
||||
[dependencies]
|
||||
tokio = { version = "1.0", default-features = false, features = ["net"] }
|
||||
bssl-tls = { path = "../bssl-tls" }
|
||||
|
||||
[dev-dependencies]
|
||||
tokio = { version = "1.0", features = ["full"] }
|
||||
futures = "0.3"
|
||||
bssl-x509 = { path = "../bssl-x509" }
|
||||
@@ -0,0 +1,497 @@
|
||||
// Copyright 2026 The BoringSSL Authors
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.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.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// 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.
|
||||
|
||||
#![deny(
|
||||
missing_docs,
|
||||
unsafe_op_in_unsafe_fn,
|
||||
clippy::missing_safety_doc,
|
||||
clippy::indexing_slicing,
|
||||
clippy::unwrap_used,
|
||||
clippy::panic,
|
||||
clippy::expect_used,
|
||||
clippy::undocumented_unsafe_blocks
|
||||
)]
|
||||
|
||||
//! `bssl-tls` / `tokio` integration
|
||||
//!
|
||||
//! This crate provides integration helpers with `tokio` and furthermore `hyper`.
|
||||
//!
|
||||
//! # Example
|
||||
//!
|
||||
//! ```rust
|
||||
//! use bssl_tls::context::TlsContextBuilder;
|
||||
//! use bssl_tls_tokio::{TokioIo, TokioTlsConnection};
|
||||
//! use tokio::net::TcpStream;
|
||||
//! # use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
//! # use tokio::net::TcpListener;
|
||||
//! # use bssl_tls::credentials::{Certificate, TlsCredentialBuilder};
|
||||
//! # use bssl_x509::{certificates::X509Certificate, keys::PrivateKey, params::Trust, store::X509StoreBuilder};
|
||||
//! #
|
||||
//! # #[tokio::main]
|
||||
//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
//! # const CA: &[u8] = include_bytes!("../../test-data/BoringSSLCATest.crt");
|
||||
//! # const RSA_SERVER_CERT: &[u8] = include_bytes!("../../test-data/BoringSSLServerTest-RSA.crt");
|
||||
//! # const RSA_SERVER_KEY: &[u8] = include_bytes!("../../test-data/BoringSSLServerTest-RSA.key");
|
||||
//! #
|
||||
//! # // Server setup
|
||||
//! # let ca = Certificate::parse_one_from_pem(CA, None)?;
|
||||
//! # let server_cert = Certificate::parse_one_from_pem(RSA_SERVER_CERT, None)?;
|
||||
//! # let server_key = PrivateKey::from_pem(RSA_SERVER_KEY, || unreachable!())?;
|
||||
//! #
|
||||
//! # let mut server_ctx_builder = TlsContextBuilder::new_tls();
|
||||
//! # let server_cred = {
|
||||
//! # let mut builder = TlsCredentialBuilder::new();
|
||||
//! # builder
|
||||
//! # .with_certificate_chain(&[server_cert, ca.clone()])?
|
||||
//! # .with_private_key(server_key)?;
|
||||
//! # builder.build()
|
||||
//! # };
|
||||
//! # server_ctx_builder.with_credential(server_cred.unwrap())?;
|
||||
//! # let server_ctx = server_ctx_builder.build();
|
||||
//! #
|
||||
//! # // Client setup
|
||||
//! # let mut client_ctx_builder = TlsContextBuilder::new_tls();
|
||||
//! # let mut cert_store = X509StoreBuilder::new();
|
||||
//! # cert_store
|
||||
//! # .set_trust(Trust::SslServer)?
|
||||
//! # .add_cert(X509Certificate::parse_one_from_pem(CA)?)?;
|
||||
//! # let cert_store = cert_store.build();
|
||||
//! # client_ctx_builder.with_certificate_store(&cert_store);
|
||||
//! #
|
||||
//! # // Connection
|
||||
//! # let listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
//! # let addr = listener.local_addr()?;
|
||||
//! #
|
||||
//! # let server_task = tokio::spawn(async move {
|
||||
//! # let (stream, _) = listener.accept().await.unwrap();
|
||||
//! # let mut conn = server_ctx.new_server_connection(None).unwrap().build();
|
||||
//! # conn.set_io(TokioIo(stream)).unwrap();
|
||||
//! # conn.in_handshake().unwrap().async_handshake().await.unwrap();
|
||||
//! #
|
||||
//! # let mut tls_stream = TokioTlsConnection::new(conn);
|
||||
//! # let mut buf = [0u8; 1024];
|
||||
//! # let _ = tls_stream.read(&mut buf).await.unwrap();
|
||||
//! # tls_stream.write_all(b"hello").await.unwrap();
|
||||
//! # tls_stream.shutdown().await.unwrap();
|
||||
//! # });
|
||||
//! #
|
||||
//! let stream = TcpStream::connect(addr).await?;
|
||||
//!
|
||||
//! let client_ctx = client_ctx_builder.build();
|
||||
//!
|
||||
//! let mut conn = client_ctx.new_client_connection(None).unwrap().build();
|
||||
//! conn.in_handshake().unwrap().set_host("www.google.com")?;
|
||||
//! conn.set_io(TokioIo(stream))?;
|
||||
//!
|
||||
//! let mut tls_stream = TokioTlsConnection::new(conn);
|
||||
//!
|
||||
//! tls_stream.write_all(b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n").await?;
|
||||
//! let mut buf = [0u8; 5];
|
||||
//! tls_stream.read_exact(&mut buf).await?;
|
||||
//! assert_eq!(&buf, b"hello");
|
||||
//! # tls_stream.shutdown().await?;
|
||||
//! # server_task.await?;
|
||||
//! # Ok(())
|
||||
//! # }
|
||||
//! ```
|
||||
|
||||
#[cfg(unix)]
|
||||
use std::os::fd::{
|
||||
AsRawFd,
|
||||
RawFd, //
|
||||
};
|
||||
use std::{
|
||||
io,
|
||||
ops::{
|
||||
Deref,
|
||||
DerefMut, //
|
||||
},
|
||||
pin::Pin,
|
||||
task::{
|
||||
Context,
|
||||
Poll,
|
||||
ready, //
|
||||
},
|
||||
};
|
||||
#[cfg(unix)]
|
||||
use tokio::io::unix::AsyncFd;
|
||||
use tokio::io::{
|
||||
AsyncRead,
|
||||
AsyncWrite,
|
||||
Interest,
|
||||
ReadBuf,
|
||||
Ready, //
|
||||
};
|
||||
|
||||
#[cfg(unix)]
|
||||
use bssl_tls::io::unix::{
|
||||
StdDatagram,
|
||||
UseFd, //
|
||||
};
|
||||
use bssl_tls::{
|
||||
connection::{
|
||||
TlsConnection,
|
||||
lifecycle::ShutdownStatus, //
|
||||
},
|
||||
context::TlsMode,
|
||||
io::{
|
||||
AbstractReader, AbstractSocket, AbstractSocketResult, AbstractWriter, IoStatus,
|
||||
NoAsyncContext, stdio::PollFor,
|
||||
}, //
|
||||
};
|
||||
|
||||
/// Translates a `std::io::Error` into an `AbstractSocketResult`.
|
||||
fn translate_stdio_err(err: io::Error) -> AbstractSocketResult {
|
||||
match err.kind() {
|
||||
io::ErrorKind::WouldBlock => AbstractSocketResult::Retry,
|
||||
io::ErrorKind::ConnectionReset
|
||||
| io::ErrorKind::ConnectionRefused
|
||||
| io::ErrorKind::ConnectionAborted
|
||||
| io::ErrorKind::BrokenPipe
|
||||
| io::ErrorKind::NotConnected
|
||||
| io::ErrorKind::UnexpectedEof => AbstractSocketResult::EndOfStream,
|
||||
_ => AbstractSocketResult::Err(Box::new(err)),
|
||||
}
|
||||
}
|
||||
|
||||
/// IO object implementing [`tokio::io::AsyncRead`] or [`tokio::io::AsyncWrite`] protocols.
|
||||
pub struct TokioIo<T>(pub T);
|
||||
|
||||
fn tokio_async_read<T: AsyncRead>(
|
||||
mut this: Pin<&mut T>,
|
||||
ctx: &mut Context<'_>,
|
||||
buffer: &mut [u8],
|
||||
) -> AbstractSocketResult {
|
||||
let mut buf = ReadBuf::new(buffer);
|
||||
loop {
|
||||
return match this.as_mut().poll_read(ctx, &mut buf) {
|
||||
Poll::Ready(Ok(())) => {
|
||||
if buf.filled().is_empty() && buf.remaining() > 0 {
|
||||
AbstractSocketResult::EndOfStream
|
||||
} else {
|
||||
AbstractSocketResult::Ok(buf.filled().len())
|
||||
}
|
||||
}
|
||||
Poll::Pending => AbstractSocketResult::Retry,
|
||||
Poll::Ready(Err(e)) => {
|
||||
if e.kind() == io::ErrorKind::Interrupted {
|
||||
continue;
|
||||
}
|
||||
translate_stdio_err(e)
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
fn tokio_async_write<T: AsyncWrite>(
|
||||
mut this: Pin<&mut T>,
|
||||
ctx: &mut Context<'_>,
|
||||
buffer: &[u8],
|
||||
) -> AbstractSocketResult {
|
||||
loop {
|
||||
return match this.as_mut().poll_write(ctx, buffer) {
|
||||
Poll::Ready(Ok(bytes)) => {
|
||||
if buffer.is_empty() {
|
||||
AbstractSocketResult::Ok(0)
|
||||
} else if bytes == 0 {
|
||||
AbstractSocketResult::EndOfStream
|
||||
} else {
|
||||
AbstractSocketResult::Ok(bytes)
|
||||
}
|
||||
}
|
||||
Poll::Pending => AbstractSocketResult::Retry,
|
||||
Poll::Ready(Err(e)) => {
|
||||
if e.kind() == io::ErrorKind::Interrupted {
|
||||
continue;
|
||||
}
|
||||
translate_stdio_err(e)
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
fn tokio_async_flush<T: AsyncWrite>(
|
||||
mut this: Pin<&mut T>,
|
||||
ctx: &mut Context<'_>,
|
||||
) -> AbstractSocketResult {
|
||||
loop {
|
||||
return match this.as_mut().poll_flush(ctx) {
|
||||
Poll::Ready(Ok(())) => AbstractSocketResult::Ok(0),
|
||||
Poll::Pending => AbstractSocketResult::Retry,
|
||||
Poll::Ready(Err(e)) => {
|
||||
if e.kind() == io::ErrorKind::Interrupted {
|
||||
continue;
|
||||
}
|
||||
translate_stdio_err(e)
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: AsyncRead + Send + Unpin> AbstractReader for TokioIo<T> {
|
||||
fn read(
|
||||
&mut self,
|
||||
async_ctx: Option<&mut Context<'_>>,
|
||||
buffer: &mut [u8],
|
||||
) -> AbstractSocketResult {
|
||||
let Some(ctx) = async_ctx else {
|
||||
return AbstractSocketResult::Err(Box::new(NoAsyncContext));
|
||||
};
|
||||
tokio_async_read(Pin::new(&mut self.0), ctx, buffer)
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: AsyncWrite + Send + Unpin> AbstractWriter for TokioIo<T> {
|
||||
fn write(
|
||||
&mut self,
|
||||
async_ctx: Option<&mut Context<'_>>,
|
||||
buffer: &[u8],
|
||||
) -> AbstractSocketResult {
|
||||
let Some(ctx) = async_ctx else {
|
||||
return AbstractSocketResult::Err(Box::new(NoAsyncContext));
|
||||
};
|
||||
tokio_async_write(Pin::new(&mut self.0), ctx, buffer)
|
||||
}
|
||||
|
||||
fn flush(&mut self, async_ctx: Option<&mut Context<'_>>) -> AbstractSocketResult {
|
||||
let Some(ctx) = async_ctx else {
|
||||
return AbstractSocketResult::Err(Box::new(NoAsyncContext));
|
||||
};
|
||||
tokio_async_flush(Pin::new(&mut self.0), ctx)
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: AsyncRead + AsyncWrite + Send + Unpin> AbstractSocket for TokioIo<T> {}
|
||||
|
||||
#[cfg(unix)]
|
||||
/// Reactor that operate over file descriptor.
|
||||
pub struct TokioOverFd(AsyncFd<RawFd>);
|
||||
|
||||
#[cfg(unix)]
|
||||
/// Construct a datagram IO object driven by [`tokio`].
|
||||
pub fn new_std_datagram_with_tokio<T: AsRawFd>(
|
||||
inner: T,
|
||||
) -> Result<StdDatagram<UseFd<T>, TokioOverFd>, io::Error> {
|
||||
let reactor = TokioOverFd::new(inner.as_raw_fd())?;
|
||||
let fd = UseFd(inner);
|
||||
Ok(StdDatagram::new(fd, reactor))
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
impl TokioOverFd {
|
||||
/// A trivial constructor to signal use of `tokio` reactor and register events with
|
||||
/// file descriptors.
|
||||
pub fn new(fd: RawFd) -> Result<Self, io::Error> {
|
||||
Ok(Self(AsyncFd::try_with_interest(
|
||||
fd,
|
||||
Interest::READABLE | Interest::WRITABLE | Interest::ERROR,
|
||||
)?))
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
impl<T> PollFor<T> for TokioOverFd {
|
||||
fn poll_read(&mut self, async_ctx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
|
||||
match ready!(self.0.poll_read_ready_mut(async_ctx)) {
|
||||
Ok(mut guard) => {
|
||||
guard.clear_ready_matching(Ready::READABLE);
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
Err(e) => Poll::Ready(Err(e)),
|
||||
}
|
||||
}
|
||||
|
||||
fn poll_write(&mut self, async_ctx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
|
||||
match ready!(self.0.poll_write_ready_mut(async_ctx)) {
|
||||
Ok(mut guard) => {
|
||||
guard.clear_ready_matching(Ready::WRITABLE);
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
Err(e) => Poll::Ready(Err(e)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Wrapper for datagram sockets to satisfy orphan rule.
|
||||
pub struct TokioDatagramIo<T>(pub T);
|
||||
|
||||
macro_rules! gen_impl_datagram {
|
||||
($ty:ty) => {
|
||||
impl AbstractReader for TokioDatagramIo<$ty> {
|
||||
fn read(
|
||||
&mut self,
|
||||
async_ctx: Option<&mut Context<'_>>,
|
||||
buffer: &mut [u8],
|
||||
) -> AbstractSocketResult {
|
||||
let Some(cx) = async_ctx else {
|
||||
return AbstractSocketResult::Err(Box::new(NoAsyncContext));
|
||||
};
|
||||
let mut buf = ReadBuf::new(buffer);
|
||||
match self.0.poll_recv(cx, &mut buf) {
|
||||
Poll::Pending => AbstractSocketResult::Retry,
|
||||
Poll::Ready(Ok(_)) => AbstractSocketResult::Ok(buf.filled().len()),
|
||||
Poll::Ready(Err(e)) => translate_stdio_err(e),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl AbstractWriter for TokioDatagramIo<$ty> {
|
||||
fn write(
|
||||
&mut self,
|
||||
async_ctx: Option<&mut Context<'_>>,
|
||||
buf: &[u8],
|
||||
) -> AbstractSocketResult {
|
||||
let Some(cx) = async_ctx else {
|
||||
return AbstractSocketResult::Err(Box::new(NoAsyncContext));
|
||||
};
|
||||
match self.0.poll_send(cx, buf) {
|
||||
Poll::Pending => AbstractSocketResult::Retry,
|
||||
Poll::Ready(Ok(len)) => AbstractSocketResult::Ok(len),
|
||||
Poll::Ready(Err(e)) => translate_stdio_err(e),
|
||||
}
|
||||
}
|
||||
|
||||
fn flush(&mut self, _: Option<&mut Context<'_>>) -> AbstractSocketResult {
|
||||
AbstractSocketResult::Ok(0)
|
||||
}
|
||||
}
|
||||
|
||||
impl AbstractSocket for TokioDatagramIo<$ty> {}
|
||||
};
|
||||
}
|
||||
|
||||
gen_impl_datagram!(tokio::net::UdpSocket);
|
||||
#[cfg(unix)]
|
||||
gen_impl_datagram!(tokio::net::UnixDatagram);
|
||||
|
||||
/// A wrapper around [`TlsConnection`] that implements Tokio's async I/O traits.
|
||||
pub struct TokioTlsConnection<Role> {
|
||||
inner: TlsConnection<Role, TlsMode>,
|
||||
}
|
||||
|
||||
impl<Role> TokioTlsConnection<Role> {
|
||||
/// Construct a new wrapper.
|
||||
pub fn new(inner: TlsConnection<Role, TlsMode>) -> Self {
|
||||
Self { inner }
|
||||
}
|
||||
|
||||
/// Consume the wrapper and return the inner connection.
|
||||
pub fn into_inner(self) -> TlsConnection<Role, TlsMode> {
|
||||
self.inner
|
||||
}
|
||||
}
|
||||
|
||||
impl<Role> Deref for TokioTlsConnection<Role> {
|
||||
type Target = TlsConnection<Role, TlsMode>;
|
||||
|
||||
fn deref(&self) -> &Self::Target {
|
||||
&self.inner
|
||||
}
|
||||
}
|
||||
|
||||
impl<Role> DerefMut for TokioTlsConnection<Role> {
|
||||
fn deref_mut(&mut self) -> &mut Self::Target {
|
||||
&mut self.inner
|
||||
}
|
||||
}
|
||||
|
||||
impl<R> AsyncRead for TokioTlsConnection<R> {
|
||||
fn poll_read(
|
||||
mut self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
buf: &mut ReadBuf<'_>,
|
||||
) -> Poll<io::Result<()>> {
|
||||
// Note: This assumes `aread_inner` is made public in `bssl-tls`.
|
||||
let status = match self
|
||||
.inner
|
||||
.as_pin_mut()
|
||||
.aread_inner(buf.initialize_unfilled(), cx)
|
||||
{
|
||||
Ok(Some(status)) => status,
|
||||
Ok(None) => return Poll::Pending,
|
||||
Err(e) => return Poll::Ready(Err(io::Error::new(io::ErrorKind::Other, e))),
|
||||
};
|
||||
match status {
|
||||
IoStatus::Ok(bytes) => {
|
||||
buf.advance(bytes);
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
IoStatus::EndOfStream => Poll::Ready(Ok(())),
|
||||
_ => Poll::Ready(Err(io::Error::new(
|
||||
io::ErrorKind::Other,
|
||||
"Unexpected I/O status",
|
||||
))),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<R> AsyncWrite for TokioTlsConnection<R> {
|
||||
fn poll_write(
|
||||
mut self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
buf: &[u8],
|
||||
) -> Poll<io::Result<usize>> {
|
||||
// Note: This assumes `awrite_inner` is made public in `bssl-tls`.
|
||||
let status = match self.inner.as_pin_mut().awrite_inner(buf, cx) {
|
||||
Ok(Some(status)) => status,
|
||||
Ok(None) => return Poll::Pending,
|
||||
Err(e) => return Poll::Ready(Err(io::Error::new(io::ErrorKind::Other, e))),
|
||||
};
|
||||
match status {
|
||||
IoStatus::Ok(bytes) => Poll::Ready(Ok(bytes)),
|
||||
IoStatus::EndOfStream => Poll::Ready(Ok(0)),
|
||||
_ => Poll::Ready(Err(io::Error::new(
|
||||
io::ErrorKind::Other,
|
||||
"Unexpected I/O status",
|
||||
))),
|
||||
}
|
||||
}
|
||||
|
||||
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
// Note: This assumes `aflush_inner` is made public in `bssl-tls`.
|
||||
let status = match self.inner.as_pin_mut().aflush_inner(cx) {
|
||||
Ok(Some(status)) => status,
|
||||
Ok(None) => return Poll::Pending,
|
||||
Err(e) => return Poll::Ready(Err(io::Error::new(io::ErrorKind::Other, e))),
|
||||
};
|
||||
match status {
|
||||
IoStatus::Ok(_) => Poll::Ready(Ok(())),
|
||||
IoStatus::EndOfStream => Poll::Ready(Ok(())),
|
||||
_ => Poll::Ready(Err(io::Error::new(
|
||||
io::ErrorKind::Other,
|
||||
"Unexpected I/O status",
|
||||
))),
|
||||
}
|
||||
}
|
||||
|
||||
fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
// Note: This assumes `ashutdown_inner` is made public in `bssl-tls`.
|
||||
match self.inner.as_pin_mut().ashutdown_inner(cx) {
|
||||
Ok(Some(ShutdownStatus::CloseNotifyReceived)) => Poll::Ready(Ok(())),
|
||||
Ok(Some(ShutdownStatus::RemainingApplicationData)) => Poll::Ready(Err(io::Error::new(
|
||||
io::ErrorKind::Other,
|
||||
"caller needs to drain application data before polling on shutdown again",
|
||||
))),
|
||||
Ok(Some(ShutdownStatus::EndOfStream)) => Poll::Ready(Err(io::Error::new(
|
||||
io::ErrorKind::UnexpectedEof,
|
||||
"unexpected eof while waiting for peek close_notify",
|
||||
))),
|
||||
Ok(Some(ShutdownStatus::CloseNotifyPosted)) => {
|
||||
unreachable!()
|
||||
}
|
||||
Ok(None) => Poll::Pending,
|
||||
Err(e) => Poll::Ready(Err(io::Error::new(io::ErrorKind::Other, e))),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,173 @@
|
||||
// Copyright 2026 The BoringSSL Authors
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.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.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// 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.
|
||||
|
||||
#![cfg(unix)]
|
||||
|
||||
use bssl_tls::{
|
||||
connection::{
|
||||
Client,
|
||||
Server,
|
||||
TlsConnection, //
|
||||
},
|
||||
context::{
|
||||
DtlsMode,
|
||||
TlsContextBuilder, //
|
||||
},
|
||||
credentials::{
|
||||
Certificate,
|
||||
TlsCredentialBuilder, //
|
||||
},
|
||||
errors::Error, //
|
||||
};
|
||||
use bssl_tls_tokio::{
|
||||
TokioDatagramIo,
|
||||
new_std_datagram_with_tokio, //
|
||||
};
|
||||
use bssl_x509::{
|
||||
certificates::X509Certificate,
|
||||
keys::PrivateKey,
|
||||
params::Trust,
|
||||
store::X509StoreBuilder, //
|
||||
};
|
||||
|
||||
const CA: &[u8] = include_bytes!("../../test-data/BoringSSLCATest.crt");
|
||||
const RSA_SERVER_CERT: &[u8] = include_bytes!("../../test-data/BoringSSLServerTest-RSA.crt");
|
||||
const RSA_SERVER_KEY: &[u8] = include_bytes!("../../test-data/BoringSSLServerTest-RSA.key");
|
||||
|
||||
fn dumb_dtls_server_client() -> Result<
|
||||
(
|
||||
TlsConnection<Server, DtlsMode>,
|
||||
TlsConnection<Client, DtlsMode>,
|
||||
),
|
||||
Error,
|
||||
> {
|
||||
let ca = Certificate::parse_one_from_pem(CA, None)?;
|
||||
let server_cert = Certificate::parse_one_from_pem(RSA_SERVER_CERT, None)?;
|
||||
let server_key = PrivateKey::from_pem(RSA_SERVER_KEY, || unreachable!())?;
|
||||
|
||||
let mut server_ctx_builder = TlsContextBuilder::new_dtls();
|
||||
let server_cred = {
|
||||
let mut builder = TlsCredentialBuilder::new();
|
||||
builder
|
||||
.with_certificate_chain(&[server_cert, ca])?
|
||||
.with_private_key(server_key)?;
|
||||
builder.build()
|
||||
};
|
||||
server_ctx_builder.with_credential(server_cred.unwrap())?;
|
||||
let server_ctx = server_ctx_builder.build();
|
||||
let server_conn = server_ctx.new_server_connection(None)?.build();
|
||||
|
||||
let mut client_ctx_builder = TlsContextBuilder::new_dtls();
|
||||
let ca = X509Certificate::parse_one_from_pem(CA)?;
|
||||
let mut cert_store = X509StoreBuilder::new();
|
||||
cert_store.set_trust(Trust::SslServer)?.add_cert(ca)?;
|
||||
let cert_store = cert_store.build();
|
||||
client_ctx_builder.with_certificate_store(&cert_store);
|
||||
let client_ctx = client_ctx_builder.build();
|
||||
let client_conn = client_ctx.new_client_connection(None)?.build();
|
||||
|
||||
Ok((server_conn, client_conn))
|
||||
}
|
||||
|
||||
async fn async_ping_pong(
|
||||
mut server_conn: TlsConnection<Server, DtlsMode>,
|
||||
mut client_conn: TlsConnection<Client, DtlsMode>,
|
||||
) -> Result<(), Error> {
|
||||
use bssl_tls::io::IoStatus;
|
||||
use std::time::Duration;
|
||||
|
||||
let task = tokio::spawn(async move {
|
||||
server_conn
|
||||
.in_handshake()
|
||||
.unwrap()
|
||||
.async_handshake()
|
||||
.await?;
|
||||
|
||||
let mut message = [0; 21];
|
||||
let mut read_bytes = 0;
|
||||
while read_bytes < 21 {
|
||||
match server_conn
|
||||
.as_pin_mut()
|
||||
.async_read(&mut message[read_bytes..])
|
||||
.await?
|
||||
{
|
||||
IoStatus::Ok(n) => read_bytes += n,
|
||||
IoStatus::EndOfStream => break,
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
assert_eq!(&message, b"BoringSSL is awesome!");
|
||||
tokio::time::sleep(Duration::from_secs(2)).await;
|
||||
server_conn
|
||||
.as_pin_mut()
|
||||
.async_write(b"Oh yeah definitely!")
|
||||
.await?;
|
||||
server_conn.as_pin_mut().async_shutdown().await?;
|
||||
Ok::<_, Error>(())
|
||||
});
|
||||
|
||||
client_conn
|
||||
.in_handshake()
|
||||
.unwrap()
|
||||
.async_handshake()
|
||||
.await?;
|
||||
client_conn
|
||||
.as_pin_mut()
|
||||
.async_write(b"BoringSSL is awesome!")
|
||||
.await?;
|
||||
let mut message = [0; 19];
|
||||
let mut read_bytes = 0;
|
||||
while read_bytes < 19 {
|
||||
match client_conn
|
||||
.as_pin_mut()
|
||||
.async_read(&mut message[read_bytes..])
|
||||
.await?
|
||||
{
|
||||
IoStatus::Ok(n) => read_bytes += n,
|
||||
IoStatus::EndOfStream => break,
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
assert_eq!(&message, b"Oh yeah definitely!");
|
||||
assert!(matches!(
|
||||
client_conn.as_pin_mut().async_shutdown().await,
|
||||
Ok(_) | Err(Error::Io(bssl_tls::errors::IoError::EndOfStream))
|
||||
));
|
||||
task.await.unwrap()?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn async_dtls() -> Result<(), Error> {
|
||||
let (mut server_conn, mut client_conn) = dumb_dtls_server_client().unwrap();
|
||||
let (server_sock, client_sock) = tokio::net::UnixDatagram::pair().unwrap();
|
||||
server_conn.set_io(TokioDatagramIo(server_sock)).unwrap();
|
||||
client_conn.set_io(TokioDatagramIo(client_sock)).unwrap();
|
||||
|
||||
async_ping_pong(server_conn, client_conn).await
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn async_dtls_over_fd() -> Result<(), Error> {
|
||||
let (mut server_conn, mut client_conn) = dumb_dtls_server_client().unwrap();
|
||||
let (server_sock, client_sock) = std::os::unix::net::UnixDatagram::pair().unwrap();
|
||||
server_sock.set_nonblocking(true).unwrap();
|
||||
client_sock.set_nonblocking(true).unwrap();
|
||||
let server_sock = new_std_datagram_with_tokio(server_sock).unwrap();
|
||||
let client_sock = new_std_datagram_with_tokio(client_sock).unwrap();
|
||||
server_conn.set_io(server_sock).unwrap();
|
||||
client_conn.set_io(client_sock).unwrap();
|
||||
|
||||
async_ping_pong(server_conn, client_conn).await
|
||||
}
|
||||
@@ -0,0 +1,128 @@
|
||||
// Copyright 2026 The BoringSSL Authors
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.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.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// 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.
|
||||
|
||||
#![cfg(unix)]
|
||||
|
||||
use bssl_tls::{
|
||||
connection::{
|
||||
Client,
|
||||
Server,
|
||||
TlsConnection, //
|
||||
},
|
||||
context::TlsContextBuilder,
|
||||
credentials::{
|
||||
Certificate,
|
||||
TlsCredentialBuilder, //
|
||||
},
|
||||
errors::Error, //
|
||||
};
|
||||
use bssl_tls_tokio::{
|
||||
TokioIo,
|
||||
TokioTlsConnection, //
|
||||
};
|
||||
use bssl_x509::{
|
||||
certificates::X509Certificate,
|
||||
keys::PrivateKey,
|
||||
params::Trust,
|
||||
store::X509StoreBuilder, //
|
||||
};
|
||||
use tokio::io::{
|
||||
AsyncReadExt,
|
||||
AsyncWriteExt, //
|
||||
};
|
||||
|
||||
use futures::future::FutureExt;
|
||||
|
||||
const CA: &[u8] = include_bytes!("../../test-data/BoringSSLCATest.crt");
|
||||
const RSA_SERVER_CERT: &[u8] = include_bytes!("../../test-data/BoringSSLServerTest-RSA.crt");
|
||||
const RSA_SERVER_KEY: &[u8] = include_bytes!("../../test-data/BoringSSLServerTest-RSA.key");
|
||||
|
||||
fn dumb_server_client() -> Result<(TlsConnection<Server>, TlsConnection<Client>), Error> {
|
||||
let ca = Certificate::parse_one_from_pem(CA, None)?;
|
||||
let server_cert = Certificate::parse_one_from_pem(RSA_SERVER_CERT, None)?;
|
||||
let server_key = PrivateKey::from_pem(RSA_SERVER_KEY, || unreachable!())?;
|
||||
|
||||
let mut server_ctx_builder = TlsContextBuilder::new_tls();
|
||||
let server_cred = {
|
||||
let mut builder = TlsCredentialBuilder::new();
|
||||
builder
|
||||
.with_certificate_chain(&[server_cert, ca])?
|
||||
.with_private_key(server_key)?;
|
||||
builder.build()
|
||||
};
|
||||
server_ctx_builder.with_credential(server_cred.unwrap())?;
|
||||
let server_ctx = server_ctx_builder.build();
|
||||
let server_conn = server_ctx.new_server_connection(None)?.build();
|
||||
|
||||
let mut client_ctx_builder = TlsContextBuilder::new_tls();
|
||||
let mut cert_store = X509StoreBuilder::new();
|
||||
cert_store
|
||||
.set_trust(Trust::SslServer)?
|
||||
.add_cert(X509Certificate::parse_one_from_pem(&CA)?)?;
|
||||
let cert_store = cert_store.build();
|
||||
client_ctx_builder.with_certificate_store(&cert_store);
|
||||
let client_ctx = client_ctx_builder.build();
|
||||
let client_conn = client_ctx.new_client_connection(None)?.build();
|
||||
|
||||
Ok((server_conn, client_conn))
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn tokio_io() -> Result<(), Error> {
|
||||
let (server_conn, client_conn) = dumb_server_client()?;
|
||||
|
||||
let (server_tx, server_rx) = tokio::net::unix::pipe::pipe().unwrap();
|
||||
let (client_tx, client_rx) = tokio::net::unix::pipe::pipe().unwrap();
|
||||
let (send_signal, recv_signal) = tokio::sync::oneshot::channel();
|
||||
let server_rx = TokioIo(server_rx);
|
||||
let server_tx = TokioIo(server_tx);
|
||||
let client_rx = TokioIo(client_rx);
|
||||
let client_tx = TokioIo(client_tx);
|
||||
|
||||
let mut server_conn = TokioTlsConnection::new(server_conn);
|
||||
let mut client_conn = TokioTlsConnection::new(client_conn);
|
||||
|
||||
server_conn.set_split_io(client_rx, server_tx)?;
|
||||
client_conn.set_split_io(server_rx, client_tx)?;
|
||||
|
||||
let thread = tokio::spawn(async move {
|
||||
let mut message = [0; 21];
|
||||
server_conn.read_exact(&mut message).await.unwrap();
|
||||
assert_eq!(message, *b"BoringSSL is awesome!");
|
||||
server_conn.write_all(b"Oh yeah definitely!").await.unwrap();
|
||||
|
||||
// The original test used sync_shutdown on established()
|
||||
// We use wrapper's shutdown instead.
|
||||
|
||||
let res = server_conn.shutdown().now_or_never();
|
||||
assert!(res.is_none(), "{res:?}");
|
||||
|
||||
// Make the client progress
|
||||
send_signal.send(()).unwrap();
|
||||
|
||||
server_conn.shutdown().await.unwrap();
|
||||
});
|
||||
client_conn
|
||||
.write_all(b"BoringSSL is awesome!")
|
||||
.await
|
||||
.unwrap();
|
||||
recv_signal.await.unwrap();
|
||||
let mut message = [0; 19];
|
||||
client_conn.read_exact(&mut message).await.unwrap();
|
||||
assert_eq!(message, *b"Oh yeah definitely!");
|
||||
client_conn.shutdown().await.unwrap();
|
||||
thread.await.unwrap();
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -20,10 +20,7 @@ path = "../bssl-sys"
|
||||
[dependencies.bssl-x509]
|
||||
path = "../bssl-x509"
|
||||
|
||||
[dependencies.tokio]
|
||||
version = "1.0"
|
||||
optional = true
|
||||
default-features = false
|
||||
|
||||
|
||||
[dependencies.once_cell]
|
||||
version = "1"
|
||||
@@ -39,18 +36,13 @@ tracing = "0.1"
|
||||
tracing-subscriber = "0.3"
|
||||
futures = "0.3"
|
||||
|
||||
[dev-dependencies.tokio]
|
||||
version = "1.0"
|
||||
features = ["full"]
|
||||
|
||||
|
||||
[features]
|
||||
default = ["std"]
|
||||
# `std` depends on the Rust `std` crate, but adds some useful trait impls if
|
||||
# available.
|
||||
std = ["bssl-crypto/std", "once_cell/std", "libc"]
|
||||
# `tokio_io` enables interop with `tokio` for async I/O operations
|
||||
tokio_io = ["dep:tokio", "std"]
|
||||
# `tokio_net` enables interop with `tokio` for async network I/O operations
|
||||
tokio_net = ["tokio_io", "tokio/net", "std"]
|
||||
|
||||
# `mlalgs` enables ML-KEM and ML-DSA support. This requires Rust 1.82.
|
||||
mlalgs = ["bssl-crypto/mlalgs"]
|
||||
|
||||
@@ -224,7 +224,8 @@ where
|
||||
self.get_connection_methods().set_pending_reason(reason);
|
||||
Ok(None)
|
||||
}
|
||||
fn aread_inner(
|
||||
#[doc(hidden)]
|
||||
pub fn aread_inner(
|
||||
self: Pin<&mut Self>,
|
||||
buffer: &mut [u8],
|
||||
cx: &mut Context<'_>,
|
||||
@@ -233,7 +234,8 @@ where
|
||||
self.do_async_io(cx, move |this| this.sync_read(&mut buffer))
|
||||
}
|
||||
|
||||
fn awrite_inner(
|
||||
#[doc(hidden)]
|
||||
pub fn awrite_inner(
|
||||
self: Pin<&mut Self>,
|
||||
buffer: &[u8],
|
||||
cx: &mut Context<'_>,
|
||||
@@ -241,7 +243,11 @@ where
|
||||
self.do_async_io(cx, move |this| this.sync_write(buffer))
|
||||
}
|
||||
|
||||
fn aflush_inner(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Result<Option<IoStatus>, Error> {
|
||||
#[doc(hidden)]
|
||||
pub fn aflush_inner(
|
||||
self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
) -> Result<Option<IoStatus>, Error> {
|
||||
self.do_async_io(cx, move |this| this.flush())
|
||||
}
|
||||
|
||||
@@ -289,7 +295,8 @@ where
|
||||
})
|
||||
}
|
||||
|
||||
fn ashutdown_inner(
|
||||
#[doc(hidden)]
|
||||
pub fn ashutdown_inner(
|
||||
mut self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
) -> Result<Option<ShutdownStatus>, Error> {
|
||||
@@ -326,6 +333,3 @@ where
|
||||
|
||||
#[cfg(feature = "std")]
|
||||
mod stdio;
|
||||
|
||||
#[cfg(feature = "tokio_io")]
|
||||
mod tokio_io_impl;
|
||||
|
||||
@@ -1,142 +0,0 @@
|
||||
// Copyright 2026 The BoringSSL Authors
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.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.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// 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.
|
||||
|
||||
use std::{
|
||||
fmt::Display,
|
||||
io,
|
||||
pin::Pin,
|
||||
task::{
|
||||
Context,
|
||||
Poll, //
|
||||
}, //
|
||||
};
|
||||
|
||||
use tokio::io::{
|
||||
AsyncRead,
|
||||
AsyncWrite,
|
||||
ReadBuf, //
|
||||
};
|
||||
|
||||
use super::{
|
||||
IoStatus,
|
||||
TlsConnectionRef,
|
||||
TlsMode, //
|
||||
};
|
||||
use crate::connection::lifecycle::ShutdownStatus;
|
||||
|
||||
#[inline]
|
||||
fn handle_io_status<T>(status: IoStatus) -> Poll<io::Result<T>> {
|
||||
match status {
|
||||
IoStatus::EndOfStream | IoStatus::Ok(_) => unreachable!(),
|
||||
IoStatus::Retry(_) => unreachable!("we should have handled retry earlier"),
|
||||
IoStatus::Err => Poll::Ready(Err(io::Error::new(
|
||||
io::ErrorKind::Other,
|
||||
"The transport has failed the I/O operation",
|
||||
))),
|
||||
IoStatus::Empty => Poll::Ready(Err(io::Error::new(
|
||||
io::ErrorKind::ConnectionReset,
|
||||
"TlsConnection has no backing transport or the transport has panicked",
|
||||
))),
|
||||
}
|
||||
}
|
||||
|
||||
impl<R> AsyncRead for TlsConnectionRef<R, TlsMode> {
|
||||
fn poll_read(
|
||||
mut self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
buf: &mut ReadBuf<'_>,
|
||||
) -> Poll<io::Result<()>> {
|
||||
let status = match self.as_mut().aread_inner(buf.initialize_unfilled(), cx) {
|
||||
Ok(status) => status,
|
||||
Err(e) => return Poll::Ready(Err(io::Error::new(io::ErrorKind::Other, e))),
|
||||
};
|
||||
match status {
|
||||
None => Poll::Pending,
|
||||
Some(IoStatus::Ok(bytes)) => {
|
||||
buf.advance(bytes);
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
Some(IoStatus::EndOfStream) => Poll::Ready(Ok(())),
|
||||
Some(status) => handle_io_status(status),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<R> AsyncWrite for TlsConnectionRef<R, TlsMode> {
|
||||
fn poll_write(
|
||||
mut self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
buf: &[u8],
|
||||
) -> Poll<io::Result<usize>> {
|
||||
let status = match self.as_mut().awrite_inner(buf, cx) {
|
||||
Ok(status) => status,
|
||||
Err(e) => return Poll::Ready(Err(io::Error::new(io::ErrorKind::Other, e))),
|
||||
};
|
||||
match status {
|
||||
None => Poll::Pending,
|
||||
Some(IoStatus::Ok(bytes)) => Poll::Ready(Ok(bytes)),
|
||||
Some(IoStatus::EndOfStream) => Poll::Ready(Ok(0)),
|
||||
Some(status) => handle_io_status(status),
|
||||
}
|
||||
}
|
||||
|
||||
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
let status = match self.as_mut().aflush_inner(cx) {
|
||||
Ok(status) => status,
|
||||
Err(e) => return Poll::Ready(Err(io::Error::new(io::ErrorKind::Other, e))),
|
||||
};
|
||||
match status {
|
||||
None => Poll::Pending,
|
||||
Some(IoStatus::Ok(_)) => Poll::Ready(Ok(())),
|
||||
Some(IoStatus::EndOfStream) => Poll::Ready(Ok(())),
|
||||
Some(status) => handle_io_status(status),
|
||||
}
|
||||
}
|
||||
|
||||
/// # Warning ⚠️
|
||||
///
|
||||
/// Calling this may fail with error [`NeedToDrainAppData`].
|
||||
/// This is not a hard error.
|
||||
/// Caller should continue reading from the connection until the end of the stream.
|
||||
fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
match self.as_mut().ashutdown_inner(cx) {
|
||||
Ok(Some(ShutdownStatus::CloseNotifyReceived)) => Poll::Ready(Ok(())),
|
||||
Ok(Some(ShutdownStatus::RemainingApplicationData)) => Poll::Ready(Err(io::Error::new(
|
||||
io::ErrorKind::Other,
|
||||
NeedToDrainAppData,
|
||||
))),
|
||||
Ok(Some(ShutdownStatus::EndOfStream)) => Poll::Ready(Err(io::Error::new(
|
||||
io::ErrorKind::UnexpectedEof,
|
||||
"unexpected eof while waiting for peek close_notify",
|
||||
))),
|
||||
Ok(Some(ShutdownStatus::CloseNotifyPosted)) => unreachable!(),
|
||||
Ok(None) => Poll::Pending,
|
||||
Err(e) => Poll::Ready(Err(io::Error::new(io::ErrorKind::Other, e))),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct NeedToDrainAppData;
|
||||
|
||||
impl Display for NeedToDrainAppData {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
write!(
|
||||
f,
|
||||
"caller needs to drain application data before polling on shutdown again"
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for NeedToDrainAppData {}
|
||||
@@ -49,9 +49,7 @@ use crate::{
|
||||
pub mod stdio;
|
||||
/// Synchronous I/O adapters.
|
||||
pub mod sync_io;
|
||||
/// Tokio-based async I/O adapters.
|
||||
#[cfg(feature = "tokio_net")]
|
||||
pub mod tokio;
|
||||
|
||||
#[cfg(all(unix, feature = "std"))]
|
||||
pub mod unix;
|
||||
|
||||
|
||||
@@ -1,131 +0,0 @@
|
||||
// Copyright 2026 The BoringSSL Authors
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.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.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// 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.
|
||||
|
||||
#![cfg(feature = "tokio_net")]
|
||||
|
||||
use std::{
|
||||
pin::Pin,
|
||||
task::{
|
||||
Context,
|
||||
Poll, //
|
||||
}, //
|
||||
};
|
||||
|
||||
use tokio::io::{
|
||||
AsyncRead,
|
||||
AsyncWrite,
|
||||
ReadBuf, //
|
||||
};
|
||||
|
||||
use crate::io::{
|
||||
AbstractReader,
|
||||
AbstractSocket,
|
||||
AbstractSocketResult,
|
||||
AbstractWriter,
|
||||
NoAsyncContext, //
|
||||
};
|
||||
|
||||
/// IO object implementing [`tokio::io::AsyncRead`] or [`tokio::io::AsyncWrite`] protocol.
|
||||
pub struct TokioIo<T>(pub T);
|
||||
|
||||
fn tokio_async_read<T: AsyncRead>(
|
||||
mut this: Pin<&mut T>,
|
||||
ctx: &mut Context<'_>,
|
||||
buffer: &mut [u8],
|
||||
) -> AbstractSocketResult {
|
||||
let mut buf = ReadBuf::new(buffer);
|
||||
loop {
|
||||
return match this.as_mut().poll_read(ctx, &mut buf) {
|
||||
Poll::Ready(Ok(())) => {
|
||||
if buf.filled().is_empty() && buf.remaining() > 0 {
|
||||
AbstractSocketResult::EndOfStream
|
||||
} else {
|
||||
AbstractSocketResult::Ok(buf.filled().len())
|
||||
}
|
||||
}
|
||||
Poll::Pending => AbstractSocketResult::Retry,
|
||||
Poll::Ready(Err(e)) => crate::retry_on_interrupt!(e),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
fn tokio_async_write<T: AsyncWrite>(
|
||||
mut this: Pin<&mut T>,
|
||||
ctx: &mut Context<'_>,
|
||||
buffer: &[u8],
|
||||
) -> AbstractSocketResult {
|
||||
loop {
|
||||
return match this.as_mut().poll_write(ctx, buffer) {
|
||||
Poll::Ready(Ok(bytes)) => {
|
||||
if buffer.is_empty() {
|
||||
AbstractSocketResult::Ok(0)
|
||||
} else if bytes == 0 {
|
||||
AbstractSocketResult::EndOfStream
|
||||
} else {
|
||||
AbstractSocketResult::Ok(bytes)
|
||||
}
|
||||
}
|
||||
Poll::Pending => AbstractSocketResult::Retry,
|
||||
Poll::Ready(Err(e)) => crate::retry_on_interrupt!(e),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
fn tokio_async_flush<T: AsyncWrite>(
|
||||
mut this: Pin<&mut T>,
|
||||
ctx: &mut Context<'_>,
|
||||
) -> AbstractSocketResult {
|
||||
loop {
|
||||
return match this.as_mut().poll_flush(ctx) {
|
||||
Poll::Ready(Ok(())) => AbstractSocketResult::Ok(0),
|
||||
Poll::Pending => AbstractSocketResult::Retry,
|
||||
Poll::Ready(Err(e)) => crate::retry_on_interrupt!(e),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: AsyncRead + Send + Unpin> AbstractReader for TokioIo<T> {
|
||||
fn read(
|
||||
&mut self,
|
||||
async_ctx: Option<&mut Context<'_>>,
|
||||
buffer: &mut [u8],
|
||||
) -> AbstractSocketResult {
|
||||
let Some(ctx) = async_ctx else {
|
||||
return AbstractSocketResult::Err(Box::new(NoAsyncContext));
|
||||
};
|
||||
tokio_async_read(Pin::new(&mut self.0), ctx, buffer)
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: AsyncWrite + Send + Unpin> AbstractWriter for TokioIo<T> {
|
||||
fn write(
|
||||
&mut self,
|
||||
async_ctx: Option<&mut Context<'_>>,
|
||||
buffer: &[u8],
|
||||
) -> AbstractSocketResult {
|
||||
let Some(ctx) = async_ctx else {
|
||||
return AbstractSocketResult::Err(Box::new(NoAsyncContext));
|
||||
};
|
||||
tokio_async_write(Pin::new(&mut self.0), ctx, buffer)
|
||||
}
|
||||
|
||||
fn flush(&mut self, async_ctx: Option<&mut Context<'_>>) -> AbstractSocketResult {
|
||||
let Some(ctx) = async_ctx else {
|
||||
return AbstractSocketResult::Err(Box::new(NoAsyncContext));
|
||||
};
|
||||
tokio_async_flush(Pin::new(&mut self.0), ctx)
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: AsyncRead + AsyncWrite + Send + Unpin> AbstractSocket for TokioIo<T> {}
|
||||
@@ -41,11 +41,6 @@ use crate::io::stdio::{
|
||||
DatagramSocket,
|
||||
PollFor, //
|
||||
};
|
||||
use crate::io::sync_io::translate_stdio_err;
|
||||
#[cfg(feature = "tokio_net")]
|
||||
pub use crate::io::tokio::TokioIo;
|
||||
#[cfg(feature = "tokio_net")]
|
||||
pub use tokio_impl::TokioOverFd;
|
||||
|
||||
use super::{
|
||||
AbstractReader,
|
||||
@@ -54,9 +49,6 @@ use super::{
|
||||
AbstractWriter, //
|
||||
};
|
||||
|
||||
#[cfg(feature = "tokio_net")]
|
||||
mod tokio_impl;
|
||||
|
||||
// ============
|
||||
// Datagrams
|
||||
// ============
|
||||
|
||||
@@ -1,106 +0,0 @@
|
||||
use std::{
|
||||
io,
|
||||
os::fd::{AsRawFd, RawFd},
|
||||
task::{Context, Poll, ready},
|
||||
};
|
||||
|
||||
use tokio::io::{Interest, ReadBuf, Ready, unix::AsyncFd};
|
||||
|
||||
use crate::io::{
|
||||
AbstractReader, AbstractSocket, AbstractSocketResult, AbstractWriter, NoAsyncContext,
|
||||
stdio::PollFor,
|
||||
unix::{StdDatagram, UseFd, translate_stdio_err},
|
||||
};
|
||||
|
||||
/// Reactor that operate over file descriptor.
|
||||
pub struct TokioOverFd(AsyncFd<RawFd>);
|
||||
|
||||
impl<T: AsRawFd> StdDatagram<UseFd<T>, TokioOverFd> {
|
||||
/// Construct a datagram IO object driven by [`tokio`].
|
||||
pub fn new_with_tokio(inner: T) -> Result<Self, io::Error> {
|
||||
let reactor = TokioOverFd::new(inner.as_raw_fd())?;
|
||||
let fd = UseFd(inner);
|
||||
Ok(Self::new(fd, reactor))
|
||||
}
|
||||
}
|
||||
|
||||
impl TokioOverFd {
|
||||
/// A trivial constructor to signal use of `tokio` reactor and register events with
|
||||
/// file descriptors.
|
||||
pub fn new(fd: RawFd) -> Result<Self, io::Error> {
|
||||
Ok(Self(AsyncFd::try_with_interest(
|
||||
fd,
|
||||
Interest::READABLE | Interest::WRITABLE | Interest::ERROR,
|
||||
)?))
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> PollFor<T> for TokioOverFd {
|
||||
fn poll_read(&mut self, async_ctx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
|
||||
match ready!(self.0.poll_read_ready_mut(async_ctx)) {
|
||||
Ok(mut guard) => {
|
||||
guard.clear_ready_matching(Ready::READABLE);
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
Err(e) => Poll::Ready(Err(e)),
|
||||
}
|
||||
}
|
||||
|
||||
fn poll_write(&mut self, async_ctx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
|
||||
match ready!(self.0.poll_write_ready_mut(async_ctx)) {
|
||||
Ok(mut guard) => {
|
||||
guard.clear_ready_matching(Ready::WRITABLE);
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
Err(e) => Poll::Ready(Err(e)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
macro_rules! gen_impl_datagram {
|
||||
($ty:ty) => {
|
||||
impl AbstractReader for $ty {
|
||||
fn read(
|
||||
&mut self,
|
||||
async_ctx: Option<&mut Context<'_>>,
|
||||
buffer: &mut [u8],
|
||||
) -> AbstractSocketResult {
|
||||
let Some(cx) = async_ctx else {
|
||||
return AbstractSocketResult::Err(Box::new(NoAsyncContext));
|
||||
};
|
||||
let mut buf = ReadBuf::new(buffer);
|
||||
match self.poll_recv(cx, &mut buf) {
|
||||
Poll::Pending => AbstractSocketResult::Retry,
|
||||
Poll::Ready(Ok(_)) => AbstractSocketResult::Ok(buf.filled().len()),
|
||||
Poll::Ready(Err(e)) => translate_stdio_err(e),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl AbstractWriter for $ty {
|
||||
fn write(
|
||||
&mut self,
|
||||
async_ctx: Option<&mut Context<'_>>,
|
||||
buf: &[u8],
|
||||
) -> AbstractSocketResult {
|
||||
let Some(cx) = async_ctx else {
|
||||
return AbstractSocketResult::Err(Box::new(NoAsyncContext));
|
||||
};
|
||||
match self.poll_send(cx, buf) {
|
||||
Poll::Pending => AbstractSocketResult::Retry,
|
||||
Poll::Ready(Ok(len)) => AbstractSocketResult::Ok(len),
|
||||
Poll::Ready(Err(e)) => translate_stdio_err(e),
|
||||
}
|
||||
}
|
||||
|
||||
fn flush(&mut self, _: Option<&mut Context<'_>>) -> AbstractSocketResult {
|
||||
AbstractSocketResult::Ok(0)
|
||||
}
|
||||
}
|
||||
|
||||
impl AbstractSocket for $ty {}
|
||||
};
|
||||
}
|
||||
|
||||
gen_impl_datagram!(tokio::net::UdpSocket);
|
||||
gen_impl_datagram!(tokio::net::UnixDatagram);
|
||||
@@ -135,63 +135,6 @@ fn sync_ping_pong<
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(feature = "tokio_io")]
|
||||
async fn async_ping_pong<
|
||||
M: crate::connection::methods::HasTlsConnectionMethod
|
||||
+ crate::context::SupportedMode
|
||||
+ crate::context::HasBasicIo
|
||||
+ 'static,
|
||||
>(
|
||||
mut server_conn: TlsConnection<Server, M>,
|
||||
mut client_conn: TlsConnection<Client, M>,
|
||||
) -> Result<(), Error> {
|
||||
use std::time::Duration;
|
||||
|
||||
let task = tokio::spawn(async move {
|
||||
server_conn
|
||||
.in_handshake()
|
||||
.unwrap()
|
||||
.async_handshake()
|
||||
.await?;
|
||||
|
||||
let mut message = [0; 21];
|
||||
assert!(matches!(
|
||||
server_conn.as_pin_mut().async_read(&mut message).await?,
|
||||
IoStatus::Ok(21)
|
||||
));
|
||||
assert_eq!(message, *b"BoringSSL is awesome!");
|
||||
tokio::time::sleep(Duration::from_secs(2)).await;
|
||||
server_conn
|
||||
.as_pin_mut()
|
||||
.async_write(b"Oh yeah definitely!")
|
||||
.await?;
|
||||
server_conn.as_pin_mut().async_shutdown().await?;
|
||||
Ok::<_, Error>(())
|
||||
});
|
||||
|
||||
client_conn
|
||||
.in_handshake()
|
||||
.unwrap()
|
||||
.async_handshake()
|
||||
.await?;
|
||||
client_conn
|
||||
.as_pin_mut()
|
||||
.async_write(b"BoringSSL is awesome!")
|
||||
.await?;
|
||||
let mut message = [0; 19];
|
||||
assert!(matches!(
|
||||
client_conn.as_pin_mut().async_read(&mut message).await?,
|
||||
IoStatus::Ok(19)
|
||||
));
|
||||
assert_eq!(message, *b"Oh yeah definitely!");
|
||||
assert!(matches!(
|
||||
client_conn.as_pin_mut().async_shutdown().await,
|
||||
Ok(_) | Err(Error::Io(crate::errors::IoError::EndOfStream))
|
||||
));
|
||||
task.await.unwrap()?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub const TEST_DATA: &[u8] = b"BoringSSL is awesome!";
|
||||
|
||||
/// A mocked pipe that will never close or emit EOF.
|
||||
|
||||
@@ -74,36 +74,6 @@ fn dtls() {
|
||||
sync_ping_pong(server_conn, client_conn).unwrap();
|
||||
}
|
||||
|
||||
#[cfg(all(unix, feature = "tokio_net"))]
|
||||
#[tokio::test]
|
||||
async fn async_dtls() -> Result<(), Error> {
|
||||
use crate::tests::async_ping_pong;
|
||||
|
||||
let (mut server_conn, mut client_conn) = dumb_dtls_server_client().unwrap();
|
||||
let (server_sock, client_sock) = tokio::net::UnixDatagram::pair().unwrap();
|
||||
server_conn.set_io(server_sock).unwrap();
|
||||
client_conn.set_io(client_sock).unwrap();
|
||||
|
||||
async_ping_pong(server_conn, client_conn).await
|
||||
}
|
||||
|
||||
#[cfg(all(feature = "tokio_net", unix))]
|
||||
#[tokio::test]
|
||||
async fn async_dtls_over_fd() -> Result<(), Error> {
|
||||
use crate::{io::unix::StdDatagram, tests::async_ping_pong};
|
||||
|
||||
let (mut server_conn, mut client_conn) = dumb_dtls_server_client().unwrap();
|
||||
let (server_sock, client_sock) = std::os::unix::net::UnixDatagram::pair().unwrap();
|
||||
server_sock.set_nonblocking(true).unwrap();
|
||||
client_sock.set_nonblocking(true).unwrap();
|
||||
let server_sock = StdDatagram::new_with_tokio(server_sock).unwrap();
|
||||
let client_sock = StdDatagram::new_with_tokio(client_sock).unwrap();
|
||||
server_conn.set_io(server_sock).unwrap();
|
||||
client_conn.set_io(client_sock).unwrap();
|
||||
|
||||
async_ping_pong(server_conn, client_conn).await
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_async_dtls() -> Result<(), Error> {
|
||||
use crate::io::IoStatus;
|
||||
|
||||
@@ -54,52 +54,3 @@ fn stdio() {
|
||||
client_conn.established().unwrap().sync_shutdown().unwrap();
|
||||
thread.join().unwrap();
|
||||
}
|
||||
|
||||
#[cfg(all(unix, feature = "tokio_net"))]
|
||||
#[tokio::test]
|
||||
async fn tokio_io() -> Result<(), crate::errors::Error> {
|
||||
use crate::io::unix::TokioIo;
|
||||
|
||||
use futures::prelude::*;
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
|
||||
let (mut server_conn, mut client_conn) = dumb_server_client()?;
|
||||
|
||||
let (server_tx, server_rx) = tokio::net::unix::pipe::pipe().unwrap();
|
||||
let (client_tx, client_rx) = tokio::net::unix::pipe::pipe().unwrap();
|
||||
let (send_signal, recv_signal) = tokio::sync::oneshot::channel();
|
||||
let server_rx = TokioIo(server_rx);
|
||||
let server_tx = TokioIo(server_tx);
|
||||
let client_rx = TokioIo(client_rx);
|
||||
let client_tx = TokioIo(client_tx);
|
||||
|
||||
server_conn.set_split_io(client_rx, server_tx)?;
|
||||
client_conn.set_split_io(server_rx, client_tx)?;
|
||||
|
||||
let thread = tokio::spawn(async move {
|
||||
let mut message = [0; 21];
|
||||
server_conn.read_exact(&mut message).await.unwrap();
|
||||
assert_eq!(message, *b"BoringSSL is awesome!");
|
||||
server_conn.write_all(b"Oh yeah definitely!").await.unwrap();
|
||||
server_conn.established().unwrap().sync_shutdown().unwrap();
|
||||
|
||||
// The following shutdown call should not make progress.
|
||||
let res = server_conn.established().unwrap().shutdown().now_or_never();
|
||||
assert!(res.is_none(), "{res:?}");
|
||||
// Make the client progress.
|
||||
send_signal.send(()).unwrap();
|
||||
server_conn.established().unwrap().shutdown().await.unwrap();
|
||||
});
|
||||
client_conn
|
||||
.write_all(b"BoringSSL is awesome!")
|
||||
.await
|
||||
.unwrap();
|
||||
recv_signal.await.unwrap();
|
||||
let mut message = [0; 19];
|
||||
client_conn.read_exact(&mut message).await.unwrap();
|
||||
assert_eq!(message, *b"Oh yeah definitely!");
|
||||
client_conn.established().unwrap().sync_shutdown().unwrap();
|
||||
thread.await.unwrap();
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user