diff --git a/tonic-examples/src/routeguide/client.rs b/tonic-examples/src/routeguide/client.rs index 34d267c..c953a36 100644 --- a/tonic-examples/src/routeguide/client.rs +++ b/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")); diff --git a/tonic-interop/Cargo.toml b/tonic-interop/Cargo.toml index e0fd778..337e427 100644 --- a/tonic-interop/Cargo.toml +++ b/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" } diff --git a/tonic-interop/data/ca.pem b/tonic-interop/data/ca.pem new file mode 100644 index 0000000..6c8511a --- /dev/null +++ b/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----- diff --git a/tonic-interop/data/server1.key b/tonic-interop/data/server1.key new file mode 100644 index 0000000..143a5b8 --- /dev/null +++ b/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----- diff --git a/tonic-interop/data/server1.pem b/tonic-interop/data/server1.pem new file mode 100644 index 0000000..f3d43fc --- /dev/null +++ b/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----- diff --git a/tonic-interop/src/bin/client.rs b/tonic-interop/src/bin/client.rs index 2854052..3b0069f 100644 --- a/tonic-interop/src/bin/client.rs +++ b/tonic-interop/src/bin/client.rs @@ -16,15 +16,17 @@ struct Opts { async fn main() -> Result<(), Box> { 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); diff --git a/tonic-interop/src/client.rs b/tonic-interop/src/client.rs index 0cece99..b8f09c7 100644 --- a/tonic-interop/src/client.rs +++ b/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; pub type UnimplementedClient = UnimplementedServiceClient; @@ -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> { - let origin = http::Uri::from_shared(format!("http://{}", addr).into()).unwrap(); - - let svc = Client::connect(origin)?; +pub async fn create(origin: http::Uri) -> Result> { + 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> { - let origin = http::Uri::from_shared(format!("http://{}", addr).into()).unwrap(); - let svc = Client::connect(origin)?; Ok(UnimplementedServiceClient::new(svc)) diff --git a/tonic/Cargo.toml b/tonic/Cargo.toml index bb21d4b..ba486a9 100644 --- a/tonic/Cargo.toml +++ b/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", +] diff --git a/tonic/src/codec/encode.rs b/tonic/src/codec/encode.rs index 49fcaa9..be32eaa 100644 --- a/tonic/src/codec/encode.rs +++ b/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>> { 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, Status>> { match self.role { diff --git a/tonic/src/server/grpc.rs b/tonic/src/server/grpc.rs index aa43943..a0639da 100644 --- a/tonic/src/server/grpc.rs +++ b/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? { diff --git a/tonic/src/service/boxed.rs b/tonic/src/service/boxed.rs index 73494ad..f32e174 100644 --- a/tonic/src/service/boxed.rs +++ b/tonic/src/service/boxed.rs @@ -24,7 +24,6 @@ where type Response = S::Response; type Error = S::Error; - // type Future = BoxFuture<'static, Result, Self::Error>>; type Future = Pin> + Send + 'static>>; diff --git a/tonic/src/service/grpc.rs b/tonic/src/service/grpc.rs new file mode 100644 index 0000000..aa28015 --- /dev/null +++ b/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 { + type ResponseBody: Body + HttpBody; + type Error: Into; + + type Future: Future, Self::Error>>; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll>; + + fn call(&mut self, request: Request) -> Self::Future; +} + +impl GrpcService for T +where + T: Service, Response = Response>, + T::Error: Into, + ResBody: Body + HttpBody, + ::Error: Into, +{ + type ResponseBody = ResBody; + type Error = T::Error; + type Future = T::Future; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + Service::poll_ready(self, cx) + } + + fn call(&mut self, request: Request) -> Self::Future { + Service::call(self, request) + } +} diff --git a/tonic/src/service/mod.rs b/tonic/src/service/mod.rs index a819efa..da899a2 100644 --- a/tonic/src/service/mod.rs +++ b/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 { - type ResponseBody: Body + HttpBody; - type Error: Into; - - type Future: Future, Self::Error>>; - - fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll>; - - fn call(&mut self, request: Request) -> Self::Future; -} - -impl GrpcService for T -where - T: Service, Response = Response>, - T::Error: Into, - ResBody: Body + HttpBody, - ::Error: Into, -{ - type ResponseBody = ResBody; - type Error = T::Error; - type Future = T::Future; - - fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { - Service::poll_ready(self, cx) - } - - fn call(&mut self, request: Request) -> Self::Future { - Service::call(self, request) - } -} +pub use self::grpc::GrpcService; diff --git a/tonic/src/transport/client.rs b/tonic/src/transport/client.rs index dd31aa1..e9b41e1 100644 --- a/tonic/src/transport/client.rs +++ b/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>(addr: Uri, ca: P) -> Result { + 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 for Client { diff --git a/tonic/src/transport/mod.rs b/tonic/src/transport/mod.rs index a7790cb..199b2d5 100644 --- a/tonic/src/transport/mod.rs +++ b/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 for Error { diff --git a/tonic/src/transport/tls.rs b/tonic/src/transport/tls.rs new file mode 100644 index 0000000..db31a09 --- /dev/null +++ b/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, +} + +impl TlsConnector { + pub async fn load>(ca: P) -> Result { + 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) -> 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 for TlsConnector { + type Response = TlsStream; + type Error = super::Error; + + type Future = + Pin> + Send + 'static>>; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + 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) + } +}