Inital tls

tonic-examples/src/routeguide/client.rs

@@ -2,7 +2,7 @@ use futures::TryStreamExt;
 use route_guide::{Point, RouteNote};
 use std::time::{Duration, Instant};
 use tokio::timer::Interval;
-use tonic::{Request, transport::Client};
+use tonic::{transport::Client, Request};
 
 mod route_guide {
     include!(concat!(env!("OUT_DIR"), "/routeguide.rs"));

tonic-interop/Cargo.toml

@@ -25,8 +25,10 @@ hyper = { git = "https://github.com/hyperium/hyper" }
 console = "0.7"
 structopt = "0.2"
 pretty_env_logger = "0.3"
-tracing-fmt = "0.0.1-alpha.1"
+
 tracing = "0.1"
+tracing-fmt = "0.0.1-alpha.1"
+tracing-log = "0.0.1-alpha.1"
 
 [build-dependencies]
 tonic-build = { path = "../tonic-build" }

tonic-interop/data/ca.pem

@@ -0,0 +1,15 @@
+-----BEGIN CERTIFICATE-----
+MIICSjCCAbOgAwIBAgIJAJHGGR4dGioHMA0GCSqGSIb3DQEBCwUAMFYxCzAJBgNV
+BAYTAkFVMRMwEQYDVQQIEwpTb21lLVN0YXRlMSEwHwYDVQQKExhJbnRlcm5ldCBX
+aWRnaXRzIFB0eSBMdGQxDzANBgNVBAMTBnRlc3RjYTAeFw0xNDExMTEyMjMxMjla
+Fw0yNDExMDgyMjMxMjlaMFYxCzAJBgNVBAYTAkFVMRMwEQYDVQQIEwpTb21lLVN0
+YXRlMSEwHwYDVQQKExhJbnRlcm5ldCBXaWRnaXRzIFB0eSBMdGQxDzANBgNVBAMT
+BnRlc3RjYTCBnzANBgkqhkiG9w0BAQEFAAOBjQAwgYkCgYEAwEDfBV5MYdlHVHJ7
++L4nxrZy7mBfAVXpOc5vMYztssUI7mL2/iYujiIXM+weZYNTEpLdjyJdu7R5gGUu
+g1jSVK/EPHfc74O7AyZU34PNIP4Sh33N+/A5YexrNgJlPY+E3GdVYi4ldWJjgkAd
+Qah2PH5ACLrIIC6tRka9hcaBlIECAwEAAaMgMB4wDAYDVR0TBAUwAwEB/zAOBgNV
+HQ8BAf8EBAMCAgQwDQYJKoZIhvcNAQELBQADgYEAHzC7jdYlzAVmddi/gdAeKPau
+sPBG/C2HCWqHzpCUHcKuvMzDVkY/MP2o6JIW2DBbY64bO/FceExhjcykgaYtCH/m
+oIU63+CFOTtR7otyQAWHqXa7q4SbCDlG7DyRFxqG0txPtGvy12lgldA2+RgcigQG
+Dfcog5wrJytaQ6UA0wE=
+-----END CERTIFICATE-----

tonic-interop/data/server1.key

@@ -0,0 +1,16 @@
+-----BEGIN PRIVATE KEY-----
+MIICdQIBADANBgkqhkiG9w0BAQEFAASCAl8wggJbAgEAAoGBAOHDFScoLCVJpYDD
+M4HYtIdV6Ake/sMNaaKdODjDMsux/4tDydlumN+fm+AjPEK5GHhGn1BgzkWF+slf
+3BxhrA/8dNsnunstVA7ZBgA/5qQxMfGAq4wHNVX77fBZOgp9VlSMVfyd9N8YwbBY
+AckOeUQadTi2X1S6OgJXgQ0m3MWhAgMBAAECgYAn7qGnM2vbjJNBm0VZCkOkTIWm
+V10okw7EPJrdL2mkre9NasghNXbE1y5zDshx5Nt3KsazKOxTT8d0Jwh/3KbaN+YY
+tTCbKGW0pXDRBhwUHRcuRzScjli8Rih5UOCiZkhefUTcRb6xIhZJuQy71tjaSy0p
+dHZRmYyBYO2YEQ8xoQJBAPrJPhMBkzmEYFtyIEqAxQ/o/A6E+E4w8i+KM7nQCK7q
+K4JXzyXVAjLfyBZWHGM2uro/fjqPggGD6QH1qXCkI4MCQQDmdKeb2TrKRh5BY1LR
+81aJGKcJ2XbcDu6wMZK4oqWbTX2KiYn9GB0woM6nSr/Y6iy1u145YzYxEV/iMwff
+DJULAkB8B2MnyzOg0pNFJqBJuH29bKCcHa8gHJzqXhNO5lAlEbMK95p/P2Wi+4Hd
+aiEIAF1BF326QJcvYKmwSmrORp85AkAlSNxRJ50OWrfMZnBgzVjDx3xG6KsFQVk2
+ol6VhqL6dFgKUORFUWBvnKSyhjJxurlPEahV6oo6+A+mPhFY8eUvAkAZQyTdupP3
+XEFQKctGz+9+gKkemDp7LBBMEMBXrGTLPhpEfcjv/7KPdnFHYmhYeBTBnuVmTVWe
+F98XJ7tIFfJq
+-----END PRIVATE KEY-----

tonic-interop/data/server1.pem

@@ -0,0 +1,16 @@
+-----BEGIN CERTIFICATE-----
+MIICnDCCAgWgAwIBAgIBBzANBgkqhkiG9w0BAQsFADBWMQswCQYDVQQGEwJBVTET
+MBEGA1UECBMKU29tZS1TdGF0ZTEhMB8GA1UEChMYSW50ZXJuZXQgV2lkZ2l0cyBQ
+dHkgTHRkMQ8wDQYDVQQDEwZ0ZXN0Y2EwHhcNMTUxMTA0MDIyMDI0WhcNMjUxMTAx
+MDIyMDI0WjBlMQswCQYDVQQGEwJVUzERMA8GA1UECBMISWxsaW5vaXMxEDAOBgNV
+BAcTB0NoaWNhZ28xFTATBgNVBAoTDEV4YW1wbGUsIENvLjEaMBgGA1UEAxQRKi50
+ZXN0Lmdvb2dsZS5jb20wgZ8wDQYJKoZIhvcNAQEBBQADgY0AMIGJAoGBAOHDFSco
+LCVJpYDDM4HYtIdV6Ake/sMNaaKdODjDMsux/4tDydlumN+fm+AjPEK5GHhGn1Bg
+zkWF+slf3BxhrA/8dNsnunstVA7ZBgA/5qQxMfGAq4wHNVX77fBZOgp9VlSMVfyd
+9N8YwbBYAckOeUQadTi2X1S6OgJXgQ0m3MWhAgMBAAGjazBpMAkGA1UdEwQCMAAw
+CwYDVR0PBAQDAgXgME8GA1UdEQRIMEaCECoudGVzdC5nb29nbGUuZnKCGHdhdGVy
+em9vaS50ZXN0Lmdvb2dsZS5iZYISKi50ZXN0LnlvdXR1YmUuY29thwTAqAEDMA0G
+CSqGSIb3DQEBCwUAA4GBAJFXVifQNub1LUP4JlnX5lXNlo8FxZ2a12AFQs+bzoJ6
+hM044EDjqyxUqSbVePK0ni3w1fHQB5rY9yYC5f8G7aqqTY1QOhoUk8ZTSTRpnkTh
+y4jjdvTZeLDVBlueZUTDRmy2feY5aZIU18vFDK08dTG0A87pppuv1LNIR3loveU8
+-----END CERTIFICATE-----

tonic-interop/src/bin/client.rs

@@ -16,15 +16,17 @@ struct Opts {
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let sub = tracing_fmt::FmtSubscriber::builder().finish();
     tracing::subscriber::set_global_default(sub).unwrap();
+    let _ = tracing_log::LogTracer::init();
 
     let matches = Opts::from_args();
 
     let test_cases = matches.test_case;
 
-    let addr = "127.0.0.1:10000".parse()?;
+    let addr = "localhost:8080";
+    let origin = http::Uri::from_shared(format!("https://{}", addr).into()).unwrap();
 
-    let mut client = client::create(addr).await?;
-    let mut unimplemented_client = client::create_unimplemented(addr).await?;
+    let mut client = client::create(origin.clone()).await?;
+    let mut unimplemented_client = client::create_unimplemented(origin).await?;
 
     for test_case in test_cases {
         println!("{:?}:", test_case);

tonic-interop/src/client.rs

@@ -1,9 +1,8 @@
 use crate::{pb::*, test_assert, TestAssertion};
 use futures_util::{future, stream, SinkExt, StreamExt};
-use std::net::SocketAddr;
 use tokio::sync::mpsc;
-use tonic::{metadata::MetadataValue, Code, Request, Response, Status};
 use tonic::transport::Client;
+use tonic::{metadata::MetadataValue, Code, Request, Response, Status};
 
 pub type TestClient = TestServiceClient<Client>;
 pub type UnimplementedClient = UnimplementedServiceClient<Client>;
@@ -22,19 +21,15 @@ const TEST_STATUS_MESSAGE: &'static str = "test status message";
 const SPECIAL_TEST_STATUS_MESSAGE: &'static str =
     "\t\ntest with whitespace\r\nand Unicode BMP ☺ and non-BMP 😈\t\n";
 
-pub async fn create(addr: SocketAddr) -> Result<TestClient, Box<dyn std::error::Error>> {
-    let origin = http::Uri::from_shared(format!("http://{}", addr).into()).unwrap();
-
-    let svc = Client::connect(origin)?;
+pub async fn create(origin: http::Uri) -> Result<TestClient, Box<dyn std::error::Error>> {
+    let svc = Client::connect_with_tls(origin, "tonic-interop/data/ca.pem").await?;
 
     Ok(TestServiceClient::new(svc))
 }
 
 pub async fn create_unimplemented(
-    addr: SocketAddr,
+    origin: http::Uri,
 ) -> Result<UnimplementedClient, Box<dyn std::error::Error>> {
-    let origin = http::Uri::from_shared(format!("http://{}", addr).into()).unwrap();
-
     let svc = Client::connect(origin)?;
 
     Ok(UnimplementedServiceClient::new(svc))

tonic/Cargo.toml

@@ -18,16 +18,24 @@ prost = "0.5"
 percent-encoding = "1.0.1"
 tower-service = "=0.3.0-alpha.1"
 tokio-codec = "=0.2.0-alpha.4"
-# async-stream = "0.1.0"
-async-stream = { path = "../../async-stream/async-stream" }
+async-stream = { git = "https://github.com/tokio-rs/async-stream" }
 http-body = "0.2.0-alpha.1"
 pin-project = "0.4.0-alpha.2"
-hyper = { git = "https://github.com/hyperium/hyper", optional = true}
 
-# tower
+# optional
+hyper = { git = "https://github.com/hyperium/hyper", optional = true}
+tokio = { version = "=0.2.0-alpha.4", default-features = false, features = ["tcp"], optional = true }
+tower-make = "=0.1.0-alpha.2"
+tokio-rustls = { path = "../../tokio-rustls", optional = true }
 tower-reconnect = { path = "../../tower/tower-reconnect", optional = true }
 tower-buffer = { path = "../../tower/tower-buffer", optional = true }
 
 [features]
 default = ["transport"]
-transport = ["hyper", "tower-reconnect", "tower-buffer"]
+transport = [
+    "hyper",
+    "tower-reconnect",
+    "tower-buffer",
+    "tokio",
+    "tokio-rustls",
+]

tonic/src/codec/encode.rs

@@ -117,7 +117,7 @@ where
     }
 
     fn poll_data(
-        self: Pin<&mut Self>,
+        mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
         let mut self_proj = self.project();
@@ -135,7 +135,7 @@ where
     }
 
     fn poll_trailers(
-        self: Pin<&mut Self>,
+        mut self: Pin<&mut Self>,
         _cx: &mut Context<'_>,
     ) -> Poll<Result<Option<HeaderMap>, Status>> {
         match self.role {

tonic/src/server/grpc.rs

@@ -136,7 +136,6 @@ where
             .await?
             .ok_or(Status::new(Code::Internal, "Missing request message."))?;
 
-
         let mut req = Request::from_http_parts(parts, message);
 
         if let Some(trailers) = stream.trailers().await? {

tonic/src/service/boxed.rs

@@ -24,7 +24,6 @@ where
     type Response = S::Response;
     type Error = S::Error;
 
-    // type Future = BoxFuture<'static, Result<Response<Self::ResponseBody>, Self::Error>>;
     type Future =
         Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
 

tonic/src/service/grpc.rs

@@ -0,0 +1,37 @@
+use crate::body::Body;
+use http::{Request, Response};
+use http_body::Body as HttpBody;
+use std::future::Future;
+use std::task::{Context, Poll};
+use tower_service::Service;
+
+pub trait GrpcService<ReqBody> {
+    type ResponseBody: Body + HttpBody;
+    type Error: Into<crate::Error>;
+
+    type Future: Future<Output = Result<Response<Self::ResponseBody>, Self::Error>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>;
+
+    fn call(&mut self, request: Request<ReqBody>) -> Self::Future;
+}
+
+impl<T, ReqBody, ResBody> GrpcService<ReqBody> for T
+where
+    T: Service<Request<ReqBody>, Response = Response<ResBody>>,
+    T::Error: Into<crate::Error>,
+    ResBody: Body + HttpBody,
+    <ResBody as HttpBody>::Error: Into<crate::Error>,
+{
+    type ResponseBody = ResBody;
+    type Error = T::Error;
+    type Future = T::Future;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Service::poll_ready(self, cx)
+    }
+
+    fn call(&mut self, request: Request<ReqBody>) -> Self::Future {
+        Service::call(self, request)
+    }
+}

tonic/src/service/mod.rs

@@ -1,44 +1,7 @@
-// TODO: make this private again
-pub mod add_origin;
+mod add_origin;
 mod boxed;
+mod grpc;
 
 pub use self::add_origin::AddOrigin;
 pub use self::boxed::BoxService;
-
-use crate::body::Body;
-use http::{Request, Response};
-use http_body::Body as HttpBody;
-use std::future::Future;
-use std::task::{Context, Poll};
-use tower_service::Service;
-
-pub trait GrpcService<ReqBody> {
-    type ResponseBody: Body + HttpBody;
-    type Error: Into<crate::Error>;
-
-    type Future: Future<Output = Result<Response<Self::ResponseBody>, Self::Error>>;
-
-    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>;
-
-    fn call(&mut self, request: Request<ReqBody>) -> Self::Future;
-}
-
-impl<T, ReqBody, ResBody> GrpcService<ReqBody> for T
-where
-    T: Service<Request<ReqBody>, Response = Response<ResBody>>,
-    T::Error: Into<crate::Error>,
-    ResBody: Body + HttpBody,
-    <ResBody as HttpBody>::Error: Into<crate::Error>,
-{
-    type ResponseBody = ResBody;
-    type Error = T::Error;
-    type Future = T::Future;
-
-    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        Service::poll_ready(self, cx)
-    }
-
-    fn call(&mut self, request: Request<ReqBody>) -> Self::Future {
-        Service::call(self, request)
-    }
-}
+pub use self::grpc::GrpcService;

tonic/src/transport/client.rs

@@ -1,3 +1,4 @@
+use super::tls::TlsConnector;
 use crate::{
     body::BoxBody,
     service::{AddOrigin, BoxService, GrpcService},
@@ -9,6 +10,7 @@ use hyper::client::connect::HttpConnector;
 use hyper::client::service::Connect;
 use hyper::{Request, Response};
 use std::future::Future;
+use std::path::Path;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use tower_buffer::{future::ResponseFuture, Buffer};
@@ -43,6 +45,22 @@ impl Client {
 
         Ok(Self { svc })
     }
+
+    pub async fn connect_with_tls<P: AsRef<Path>>(addr: Uri, ca: P) -> Result<Self, super::Error> {
+        let settings = Builder::new().http2_only(true).clone();
+
+        let tls_connector = TlsConnector::load(ca).await?;
+
+        let maker = Connect::new(tls_connector, settings);
+        let svc = tower_reconnect::Reconnect::new(maker, addr.clone());
+
+        let svc = AddOrigin::new(svc, addr);
+        let svc = BoxService::new(svc);
+
+        let svc = Buffer::new(Box::new(svc) as Inner, 100);
+
+        Ok(Self { svc })
+    }
 }
 
 impl GrpcService<BoxBody> for Client {

tonic/src/transport/mod.rs

@@ -1,4 +1,5 @@
 mod client;
+mod tls;
 
 pub use self::client::Client;
 
@@ -12,7 +13,7 @@ pub struct Error {
 #[derive(Debug)]
 pub(crate) enum ErrorKind {
     Client,
-    // Server,
+    UnableToNegotiateH2, // Server,
 }
 
 impl From<ErrorKind> for Error {

tonic/src/transport/tls.rs

@@ -0,0 +1,95 @@
+use http::Uri;
+use hyper::client::connect::HttpConnector;
+use std::{
+    future::Future,
+    path::Path,
+    pin::Pin,
+    sync::Arc,
+    task::{Context, Poll},
+};
+use tokio::{fs, net::TcpStream};
+use tokio_rustls::{
+    client::TlsStream, rustls::{ClientConfig, Session}, webpki::DNSNameRef, TlsConnector as RustlsConnector,
+};
+use tower_make::MakeConnection;
+use tower_service::Service;
+
+const ALPN_H2: &str = "h2";
+
+#[derive(Clone)]
+pub struct TlsConnector {
+    http: HttpConnector,
+    config: Arc<ClientConfig>,
+}
+
+impl TlsConnector {
+    pub async fn load<P: AsRef<Path>>(ca: P) -> Result<Self, super::Error> {
+        let pem = fs::read(ca).await.map_err(|e| super::Error::from((super::ErrorKind::Client, e.into())))?;
+        Ok(TlsConnector::new(pem))
+    }
+
+    fn new(ca: Vec<u8>) -> Self {
+        let mut buf = std::io::Cursor::new(ca);
+
+        let mut config = ClientConfig::new();
+
+        config.root_store.add_pem_file(&mut buf).unwrap();
+        config.set_protocols(&[Vec::from(&ALPN_H2[..])]);
+
+        let mut http = HttpConnector::new();
+        http.enforce_http(false);
+
+        Self {
+            http,
+            config: Arc::new(config),
+        }
+    }
+}
+
+impl Service<Uri> for TlsConnector {
+    type Response = TlsStream<TcpStream>;
+    type Error = super::Error;
+
+    type Future =
+        Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        MakeConnection::poll_ready(&mut self.http, cx).map_err(|e| super::Error::from((super::ErrorKind::Client, e.into())))
+    }
+
+    fn call(&mut self, uri: Uri) -> Self::Future {
+        let auth = uri.authority_part().unwrap();
+        let dns = DNSNameRef::try_from_ascii_str("foo.test.google.fr")//auth.host())
+            .unwrap()
+            .to_owned();
+        let config = self.config.clone();
+        let connect = self.http.make_connection(uri.clone());
+
+        let fut = async move {
+            let io = match connect.await {
+                Ok(io) => io,
+                Err(e) => return Err(super::Error::from((super::ErrorKind::Client, e.into()))),
+            };
+
+            RustlsConnector::from(config)
+                .connect(dns.as_ref(), io)
+                .await
+                .map_err(|e| {
+                    println!("TLS ERROR={:?}", e);
+                    super::Error::from((super::ErrorKind::Client, e.into()))
+                })
+                .and_then(|conn| {
+                    let (_, session) = conn.get_ref();
+                    let negotiated_protocol = session.get_alpn_protocol();
+
+                    if Some(ALPN_H2.as_bytes()) == negotiated_protocol.as_ref().map(|x| &**x) {
+                        Ok(conn)
+                    } else {
+                        Err(super::Error::from(super::ErrorKind::UnableToNegotiateH2).into())
+                    }
+                })
+        };
+
+        Box::pin(fut)
+    }
+}