Add tokio-postgres-native-tls

Cargo.toml

@@ -7,5 +7,6 @@ members = [
     "postgres-openssl",
     "postgres-native-tls",
     "tokio-postgres",
+    "tokio-postgres-native-tls",
     "tokio-postgres-openssl",
 ]

tokio-postgres-native-tls/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "tokio-postgres-native-tls"
+version = "0.1.0"
+authors = ["Steven Fackler <sfackler@gmail.com>"]
+
+[dependencies]
+bytes = "0.4"
+futures = "0.1"
+native-tls = "0.2"
+tokio-io = "0.1"
+tokio-tls = "0.2"
+tokio-postgres = { version = "0.3", path = "../tokio-postgres" }
+
+[dev-dependencies]
+tokio = "0.1.7"

tokio-postgres-native-tls/src/lib.rs

@@ -0,0 +1,106 @@
+extern crate bytes;
+extern crate futures;
+extern crate native_tls;
+extern crate tokio_io;
+extern crate tokio_postgres;
+extern crate tokio_tls;
+
+#[cfg(test)]
+extern crate tokio;
+
+use bytes::{Buf, BufMut};
+use futures::{Future, Poll};
+use std::error::Error;
+use std::io::{self, Read, Write};
+use tokio_io::{AsyncRead, AsyncWrite};
+use tokio_postgres::tls::{Socket, TlsConnect, TlsStream};
+
+#[cfg(test)]
+mod test;
+
+pub struct TlsConnector {
+    connector: tokio_tls::TlsConnector,
+}
+
+impl TlsConnector {
+    pub fn new() -> Result<TlsConnector, native_tls::Error> {
+        let connector = native_tls::TlsConnector::new()?;
+        Ok(TlsConnector::with_connector(connector))
+    }
+
+    pub fn with_connector(connector: native_tls::TlsConnector) -> TlsConnector {
+        TlsConnector {
+            connector: tokio_tls::TlsConnector::from(connector),
+        }
+    }
+}
+
+impl TlsConnect for TlsConnector {
+    fn connect(
+        &self,
+        domain: &str,
+        socket: Socket,
+    ) -> Box<Future<Item = Box<TlsStream>, Error = Box<Error + Sync + Send>> + Sync + Send> {
+        let f = self
+            .connector
+            .connect(domain, socket)
+            .map(|s| {
+                let s: Box<TlsStream> = Box::new(SslStream(s));
+                s
+            }).map_err(|e| {
+                let e: Box<Error + Sync + Send> = Box::new(e);
+                e
+            });
+        Box::new(f)
+    }
+}
+
+struct SslStream(tokio_tls::TlsStream<Socket>);
+
+impl Read for SslStream {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        self.0.read(buf)
+    }
+}
+
+impl AsyncRead for SslStream {
+    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
+        self.0.prepare_uninitialized_buffer(buf)
+    }
+
+    fn read_buf<B>(&mut self, buf: &mut B) -> Poll<usize, io::Error>
+    where
+        B: BufMut,
+    {
+        self.0.read_buf(buf)
+    }
+}
+
+impl Write for SslStream {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.0.write(buf)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        self.0.flush()
+    }
+}
+
+impl AsyncWrite for SslStream {
+    fn shutdown(&mut self) -> Poll<(), io::Error> {
+        self.0.shutdown()
+    }
+
+    fn write_buf<B>(&mut self, buf: &mut B) -> Poll<usize, io::Error>
+    where
+        B: Buf,
+    {
+        self.0.write_buf(buf)
+    }
+}
+
+impl TlsStream for SslStream {
+    fn tls_server_end_point(&self) -> Option<Vec<u8>> {
+        self.0.get_ref().tls_server_end_point().unwrap_or(None)
+    }
+}

tokio-postgres-native-tls/src/test.rs

@@ -0,0 +1,69 @@
+use futures::{Future, Stream};
+use native_tls::{self, Certificate};
+use tokio::runtime::current_thread::Runtime;
+use tokio_postgres::{self, TlsMode};
+
+use TlsConnector;
+
+fn smoke_test(url: &str, tls: TlsMode) {
+    let mut runtime = Runtime::new().unwrap();
+
+    let handshake = tokio_postgres::connect(url.parse().unwrap(), tls);
+    let (mut client, connection) = runtime.block_on(handshake).unwrap();
+    let connection = connection.map_err(|e| panic!("{}", e));
+    runtime.handle().spawn(connection).unwrap();
+
+    let prepare = client.prepare("SELECT 1::INT4");
+    let statement = runtime.block_on(prepare).unwrap();
+    let select = client.query(&statement, &[]).collect().map(|rows| {
+        assert_eq!(rows.len(), 1);
+        assert_eq!(rows[0].get::<_, i32>(0), 1);
+    });
+    runtime.block_on(select).unwrap();
+
+    drop(statement);
+    drop(client);
+    runtime.run().unwrap();
+}
+
+#[test]
+fn require() {
+    let connector = native_tls::TlsConnector::builder()
+        .add_root_certificate(
+            Certificate::from_pem(include_bytes!("../../test/server.crt")).unwrap(),
+        ).build()
+        .unwrap();
+    let connector = TlsConnector::with_connector(connector);
+    smoke_test(
+        "postgres://ssl_user@localhost:5433/postgres",
+        TlsMode::Require(Box::new(connector)),
+    );
+}
+
+#[test]
+fn prefer() {
+    let connector = native_tls::TlsConnector::builder()
+        .add_root_certificate(
+            Certificate::from_pem(include_bytes!("../../test/server.crt")).unwrap(),
+        ).build()
+        .unwrap();
+    let connector = TlsConnector::with_connector(connector);
+    smoke_test(
+        "postgres://ssl_user@localhost:5433/postgres",
+        TlsMode::Prefer(Box::new(connector)),
+    );
+}
+
+#[test]
+fn scram_user() {
+    let connector = native_tls::TlsConnector::builder()
+        .add_root_certificate(
+            Certificate::from_pem(include_bytes!("../../test/server.crt")).unwrap(),
+        ).build()
+        .unwrap();
+    let connector = TlsConnector::with_connector(connector);
+    smoke_test(
+        "postgres://scram_user:password@localhost:5433/postgres",
+        TlsMode::Require(Box::new(connector)),
+    );
+}