tokio based async support

Cargo.toml

@@ -16,6 +16,11 @@ openssl = "0.10.19"
 openssl-probe = "0.1.2"
 log = "0.4.6"
 bytes = "0.4.11"
+futures = {version = "0.3", optional = true}
+tokio = {version = "0.2", features = ["default", "udp"], optional = true}
+tokio-openssl = {version="0.4", optional = true}
 
 [features]
 vendored = ["openssl/vendored"]
+async = ["futures", "tokio", "tokio-openssl"]
+

src/dtls_acceptor.rs

@@ -1,5 +1,5 @@
 use crate::openssl::try_set_supported_protocols;
-use crate::{DtlsAcceptorBuilder, DtlsStream, HandshakeError, CertificateIdentity, Protocol, Result};
+use crate::{DtlsAcceptorBuilder, SyncDtlsStream, HandshakeError, CertificateIdentity, Protocol, Result};
 use openssl::ssl::{SslAcceptor, SslMethod};
 use std::{fmt, io, result};
 
@@ -75,12 +75,12 @@ impl DtlsAcceptor {
     pub fn accept<S: fmt::Debug>(
         &self,
         stream: S,
-    ) -> result::Result<DtlsStream<S>, HandshakeError<S>>
+    ) -> result::Result<SyncDtlsStream<S>, HandshakeError<S>>
     where
         S: io::Read + io::Write,
     {
         let stream = self.0.accept(stream)?;
-        Ok(DtlsStream::from(stream))
+        Ok(SyncDtlsStream::from(stream))
     }
 }
 

src/dtls_connector.rs

@@ -1,12 +1,19 @@
 use crate::{
     openssl::{init_trust, try_set_supported_protocols},
-    DtlsConnectorBuilder, DtlsStream, Error, HandshakeError, Protocol, ConnectorIdentity
+    DtlsConnectorBuilder, SyncDtlsStream, Error, Protocol, ConnectorIdentity
 };
 use log::debug;
-use openssl::ssl::{SslConnector, SslMethod, SslVerifyMode};
+use openssl::ssl::{SslConnector, SslMethod, SslVerifyMode, ConnectConfiguration, HandshakeError};
 use openssl::error::ErrorStack;
 use std::{fmt, io, io::Write};
 
+#[cfg(feature="async")]
+use crate::AsyncDtlsStream;
+#[cfg(feature="async")]
+use tokio_openssl;
+#[cfg(feature="async")]
+use tokio::io::{AsyncRead, AsyncWrite};
+
 /// Connector to an UDP endpoint secured with DTLS.
 #[derive(Clone)]
 pub struct DtlsConnector {
@@ -21,7 +28,7 @@ impl DtlsConnector {
     ///
     /// The `DtlsConnector` will use the settings from the given builder.
     ///
-    /// The following propperties will be applied from the builder:
+    /// The following properties will be applied from the builder:
     /// - Sets minimal/maximal protocol version
     /// - Sets srtp profile by enabling the DTLS extension 'use_srtp'
     /// - Sets the certificate and private key
@@ -109,6 +116,20 @@ impl DtlsConnector {
         }
     }
 
+    fn configure(&self) -> Result<ConnectConfiguration, ErrorStack> {
+        let mut ssl = self
+            .connector
+            .configure()?
+            .use_server_name_indication(self.use_sni)
+            .verify_hostname(!self.accept_invalid_hostnames);
+
+        if self.accept_invalid_certs {
+            ssl.set_verify(SslVerifyMode::NONE);
+        }
+
+        Ok(ssl)
+    }
+
     /// Initiates a DTLS handshake.
     ///
     /// The provided domain will be used for both SNI and certificate hostname
@@ -125,22 +146,37 @@ impl DtlsConnector {
         &self,
         domain: &str,
         stream: S,
-    ) -> Result<DtlsStream<S>, HandshakeError<S>>
+    ) -> Result<SyncDtlsStream<S>, HandshakeError<S>>
     where
         S: io::Read + io::Write,
     {
-        let mut ssl = self
-            .connector
-            .configure()?
-            .use_server_name_indication(self.use_sni)
-            .verify_hostname(!self.accept_invalid_hostnames);
-        if self.accept_invalid_certs {
-            ssl.set_verify(SslVerifyMode::NONE);
-        }
+        let conf = self.configure()?;
+        let stream = conf.connect(domain, stream)?;
 
-        let stream = ssl.connect(domain, stream)?;
-        Ok(DtlsStream::from(stream))
+        Ok(SyncDtlsStream::from(stream))
     }
+
+    #[cfg(feature="async")]
+    pub async fn async_connect<S: fmt::Debug>(
+        &self,
+        domain: &str,
+        stream: S,
+    ) -> Result<AsyncDtlsStream<S>, AsyncConnectError<S>>
+    where
+        S: AsyncRead + AsyncWrite + Unpin
+    {
+        let conf = self.configure().map_err(AsyncConnectError::ErrorStack)?;
+        let stream = tokio_openssl::connect(conf, domain, stream).await.map_err(AsyncConnectError::TokioOpenSsl)?;
+
+        Ok(AsyncDtlsStream::from(stream))
+    }
+}
+
+#[cfg(feature="async")]
+#[derive(Debug)]
+pub enum AsyncConnectError<S> {
+    ErrorStack(ErrorStack),
+    TokioOpenSsl(tokio_openssl::HandshakeError<S>)
 }
 
 impl AsRef<SslConnector> for DtlsConnector {

src/dtls_stream.rs

@@ -1,6 +1,6 @@
 use crate::{Certificate, Error, SrtpProfile};
 use openssl::ssl;
-use openssl::ssl::SslStream;
+use openssl::ssl::{SslStream, SslRef};
 use std::{fmt, io};
 
 /// A stream managing a DTLS session.
@@ -9,81 +9,33 @@ use std::{fmt, io};
 /// and both the server and the client are ready for receiving and sending
 /// data. Bytes read from a `DtlsStream` are decrypted from `S` and bytes written
 /// to a `DtlsStream` are encrypted when passing through to `S`.
-pub struct DtlsStream<S>(ssl::SslStream<S>);
+pub struct DtlsStream<C>(pub(crate) C);
 
-impl<S: fmt::Debug> fmt::Debug for DtlsStream<S> {
+pub type SyncDtlsStream<S> = DtlsStream<SslStream<S>>;
+
+impl<S: fmt::Debug> fmt::Debug for DtlsStream<SslStream<S>> {
     fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
         fmt::Debug::fmt(&self.0, fmt)
     }
 }
 
-impl<S: io::Read + io::Write> DtlsStream<S> {
-    /// Export keying material
-    ///
-    /// # Underlying SSL
-    /// This corresponds to [`SSL_export_keying_material`].
-    ///
-    /// [`SSL_export_keying_material`]: https://www.openssl.org/docs/manmaster/man3/SSL_export_keying_material.html
-    pub fn keying_material(&self, len: usize) -> Result<Vec<u8>, Error> {
-        let mut buf = vec![0; len];
-        self.0
-            .ssl()
-            .export_keying_material(&mut buf, "EXTRACTOR-dtls_srtp", None)?;
-        Ok(buf)
-    }
-
-    /// Gets the SRTP profile selected by handshake.
-    ///
-    /// # Underlying SSL
-    /// DTLS extension "use_srtp" as defined in RFC5764 has to be enabled.
-    ///
-    /// This corresponds to [`SSL_get_selected_srtp_profile`].
-    ///
-    /// [`SSL_get_selected_srtp_profile`]: https://www.openssl.org/docs/man1.1.1/man3/SSL_CTX_set_tlsext_use_srtp.html
-    pub fn selected_srtp_profile(&self) -> Result<Option<SrtpProfile>, Error> {
-        match self.0.ssl().selected_srtp_profile() {
-            Some(profile) => Ok(profile.name().parse()?).map(Some),
-            None => Ok(None),
-        }
+impl<S> DtlsStreamExt<S> for SyncDtlsStream<S> {
+    fn ssl(&self) -> &SslRef {
+        &self.0.ssl()
     }
 
     /// Returns a shared reference to the underlying stream.
-    pub fn get_ref(&self) -> &S {
+    fn get_ref(&self) -> &S {
         self.0.get_ref()
     }
 
     /// Returns a mutable reference to the inner stream.
-    pub fn get_mut(&mut self) -> &mut S {
+    fn get_mut(&mut self) -> &mut S {
         self.0.get_mut()
     }
+}
 
-    /// Returns the number of bytes remaining in the currently processed TLS record.
-    ///
-    /// If this is greater than 0, the next call to `read` will not call down to the underlying
-    /// stream.
-    ///
-    /// # Underlying SSL
-    /// This corresponds to [`SSL_pending`].
-    ///
-    /// [`SSL_pending`]: https://www.openssl.org/docs/man1.1.0/ssl/SSL_pending.html
-    pub fn buffered_read_size(&self) -> Result<usize, Error> {
-        Ok(self.0.ssl().pending())
-    }
-
-    /// Returns the peer's certificate, if present.
-    ///
-    /// # Underlying SSL
-    /// This corresponds to [`SSL_get_peer_certificate`].
-    ///
-    /// [`SSL_get_peer_certificate`]: https://www.openssl.org/docs/man1.1.0/ssl/SSL_get_peer_certificate.html
-    pub fn peer_certificate(&self) -> Result<Option<Certificate>, Error> {
-        Ok(self
-            .0
-            .ssl()
-            .peer_certificate()
-            .map(|c| Certificate::from(c)))
-    }
-
+impl<S: io::Read + io::Write> SyncDtlsStream<S> {
     /// Shuts down the session.
     ///
     /// The shutdown process consists of two steps. The first step sends a close notify message to
@@ -101,6 +53,7 @@ impl<S: io::Read + io::Write> DtlsStream<S> {
     pub fn shutdown(&mut self) -> io::Result<()> {
         match self.0.shutdown() {
             Ok(_) => Ok(()),
+
             Err(ref e) if e.code() == ssl::ErrorCode::ZERO_RETURN => Ok(()),
             Err(e) => Err(e
                 .into_io_error()
@@ -109,13 +62,13 @@ impl<S: io::Read + io::Write> DtlsStream<S> {
     }
 }
 
-impl<S: io::Read + io::Write> io::Read for DtlsStream<S> {
+impl<S: io::Read + io::Write> io::Read for SyncDtlsStream<S> {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         self.0.read(buf)
     }
 }
 
-impl<S: io::Read + io::Write> io::Write for DtlsStream<S> {
+impl<S: io::Read + io::Write> io::Write for SyncDtlsStream<S> {
     fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
         self.0.write(buf)
     }
@@ -125,14 +78,77 @@ impl<S: io::Read + io::Write> io::Write for DtlsStream<S> {
     }
 }
 
-impl<S: io::Read + io::Write> AsRef<SslStream<S>> for DtlsStream<S> {
+impl<S: io::Read + io::Write> AsRef<SslStream<S>> for SyncDtlsStream<S> {
     fn as_ref(&self) -> &SslStream<S> {
         &self.0
     }
 }
 
-impl<S: io::Read + io::Write> From<SslStream<S>> for DtlsStream<S> {
+impl<S: io::Read + io::Write> From<SslStream<S>> for SyncDtlsStream<S> {
     fn from(stream: SslStream<S>) -> Self {
         DtlsStream(stream)
     }
 }
+
+
+pub trait DtlsStreamExt<S> {
+    fn ssl(&self) -> &SslRef;
+    fn get_ref(&self) -> &S;
+    fn get_mut(&mut self) -> &mut S;
+
+    /// Export keying material
+    ///
+    /// # Underlying SSL
+    /// This corresponds to [`SSL_export_keying_material`].
+    ///
+    /// [`SSL_export_keying_material`]: https://www.openssl.org/docs/manmaster/man3/SSL_export_keying_material.html
+    fn keying_material(&self, len: usize) -> Result<Vec<u8>, Error> {
+        let mut buf = vec![0; len];
+        self
+            .ssl()
+            .export_keying_material(&mut buf, "EXTRACTOR-dtls_srtp", None)?;
+        Ok(buf)
+    }
+
+    /// Gets the SRTP profile selected by handshake.
+    ///
+    /// # Underlying SSL
+    /// DTLS extension "use_srtp" as defined in RFC5764 has to be enabled.
+    ///
+    /// This corresponds to [`SSL_get_selected_srtp_profile`].
+    ///
+    /// [`SSL_get_selected_srtp_profile`]: https://www.openssl.org/docs/man1.1.1/man3/SSL_CTX_set_tlsext_use_srtp.html
+    fn selected_srtp_profile(&self) -> Result<Option<SrtpProfile>, Error> {
+        match self.ssl().selected_srtp_profile() {
+            Some(profile) => Ok(profile.name().parse()?).map(Some),
+            None => Ok(None),
+        }
+    }
+
+
+    /// Returns the number of bytes remaining in the currently processed TLS record.
+    ///
+    /// If this is greater than 0, the next call to `read` will not call down to the underlying
+    /// stream.
+    ///
+    /// # Underlying SSL
+    /// This corresponds to [`SSL_pending`].
+    ///
+    /// [`SSL_pending`]: https://www.openssl.org/docs/man1.1.0/ssl/SSL_pending.html
+    fn buffered_read_size(&self) -> Result<usize, Error> {
+        Ok(self.ssl().pending())
+    }
+
+    /// Returns the peer's certificate, if present.
+    ///
+    /// # Underlying SSL
+    /// This corresponds to [`SSL_get_peer_certificate`].
+    ///
+    /// [`SSL_get_peer_certificate`]: https://www.openssl.org/docs/man1.1.0/ssl/SSL_get_peer_certificate.html
+    fn peer_certificate(&self) -> Result<Option<Certificate>, Error> {
+        Ok(self
+           .ssl()
+           .peer_certificate()
+           .map(|c| Certificate::from(c)))
+    }
+}