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"]
+

examples/udp_socket.rs

@@ -6,12 +6,12 @@ use std::{
     time::Duration,
 };
 
-use udp_dtls::{Certificate, DtlsAcceptor, DtlsConnector, Identity, SrtpProfile};
+use udp_dtls::{Certificate, DtlsAcceptor, DtlsConnector, CertificateIdentity, SrtpProfile};
 use udp_dtls::{DtlsAcceptorBuilder, UdpChannel};
 
 fn main() {
     let buffer = include_bytes!("../test/identity.p12");
-    let identity = Identity::from_pkcs12(buffer, "mypass").unwrap();
+    let identity = CertificateIdentity::from_pkcs12(buffer, "mypass").unwrap();
 
     let root_ca = include_bytes!("../test/root-ca.der");
     let root_ca = Certificate::from_der(root_ca).unwrap();

examples/udp_socket_async.rs

@@ -0,0 +1,74 @@
+use std::time::Duration;
+use futures::pin_mut;
+use tokio::{net, time, io::{AsyncReadExt, AsyncWriteExt}};
+use udp_dtls::{Certificate, DtlsAcceptor, DtlsConnector, CertificateIdentity, SrtpProfile, AsyncUdpChannel};
+
+#[tokio::main]
+async fn main() {
+    let buffer = include_bytes!("../test/identity.p12");
+    let identity = CertificateIdentity::from_pkcs12(buffer, "mypass").unwrap();
+
+    let root_ca = include_bytes!("../test/root-ca.der");
+    let root_ca = Certificate::from_der(root_ca).unwrap();
+
+    let acceptor = DtlsAcceptor::builder(identity).build().unwrap();
+    let connector = DtlsConnector::builder()
+        .add_srtp_profile(SrtpProfile::Aes128CmSha180)
+        .add_srtp_profile(SrtpProfile::AeadAes256Gcm)
+        .add_root_certificate(root_ca)
+        .build()
+        .unwrap();
+
+    let server = net::UdpSocket::bind("127.0.0.1:0").await.unwrap();
+    let client = net::UdpSocket::bind("127.0.0.1:0").await.unwrap();
+
+    let server_addr = server.local_addr().unwrap();
+    let client_addr = client.local_addr().unwrap();
+
+    let server_channel = AsyncUdpChannel {
+        socket: server,
+        remote_addr: client_addr,
+    };
+
+    let client_channel = AsyncUdpChannel {
+        socket: client,
+        remote_addr: server_addr,
+    };
+
+    let connector_task = async move {
+        let stream = connector.async_connect("foobar.com", client_channel).await.unwrap();
+
+        pin_mut!(stream);
+        loop {
+            let buf = b"hello";
+            stream.write_all(buf).await.unwrap();
+
+            time::delay_for(Duration::from_millis(30)).await;
+        }
+    };
+
+    let acceptor_task = async move {
+        let stream = acceptor.async_accept(server_channel).await.unwrap();
+        let mut count = 0;
+
+        pin_mut!(stream);
+        loop {
+            let mut received = [0; 5];
+
+            stream.read_exact(&mut received).await.unwrap();
+
+            println!(
+                "{:?} {:?}",
+                count,
+                String::from_utf8_lossy(received.as_ref())
+            );
+
+            count = count + 1;
+
+            time::delay_for(Duration::from_millis(2)).await;
+        }
+    };
+
+    tokio::spawn(connector_task);
+    tokio::spawn(acceptor_task).await.unwrap();
+}

src/dtls_acceptor.rs

@@ -1,8 +1,15 @@
 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};
 
+#[cfg(feature="async")]
+use crate::{AsyncDtlsStream, AsyncConnectError};
+#[cfg(feature="async")]
+use tokio_openssl;
+#[cfg(feature="async")]
+use tokio::io::{AsyncRead, AsyncWrite};
+
 /// Acceptor for incoming UDP sessions secured with DTLS.
 #[derive(Clone)]
 pub struct DtlsAcceptor(SslAcceptor);
@@ -75,12 +82,25 @@ 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))
+    }
+
+    #[cfg(feature="async")]
+    pub async fn async_accept<S: fmt::Debug>(
+        &self,
+        stream: S
+    ) -> std::result::Result<AsyncDtlsStream<S>, AsyncConnectError<S>>
+    where
+        S: AsyncRead + AsyncWrite + Unpin
+    {
+        let stream = tokio_openssl::accept(&self.0, stream).await.map_err(AsyncConnectError::TokioOpenSsl)?;
+
+        Ok(AsyncDtlsStream::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, AsyncConnectError};
+#[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,21 +146,29 @@ 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)?;
+
+        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)?;
 
-        let stream = ssl.connect(domain, stream)?;
-        Ok(DtlsStream::from(stream))
+        Ok(AsyncDtlsStream::from(stream))
     }
 }