From b82439af17b8bca6c50ba679299fa57753b0be00 Mon Sep 17 00:00:00 2001 From: Lucio Franco Date: Mon, 2 Sep 2019 18:11:36 -0400 Subject: [PATCH] Switch to using endpoint --- tonic-examples/src/helloworld/client.rs | 5 +- tonic/Cargo.toml | 4 +- tonic/src/transport/channel.rs | 69 +++++++---------------- tonic/src/transport/endpoint.rs | 33 +++++++++++ tonic/src/transport/mod.rs | 7 +-- tonic/src/transport/openssl.rs | 72 ------------------------ tonic/src/transport/service/connect.rs | 13 +++-- tonic/src/transport/service/connector.rs | 36 ++++++++---- tonic/src/transport/service/discover.rs | 17 +++--- tonic/src/transport/service/mod.rs | 2 +- tonic/src/transport/service/tls/mod.rs | 1 - tonic/src/transport/tls/mod.rs | 33 +++++++++++ tonic/src/transport/tls/openssl.rs | 42 ++++++++++++++ tonic/src/transport/{ => tls}/rustls.rs | 0 14 files changed, 180 insertions(+), 154 deletions(-) create mode 100644 tonic/src/transport/endpoint.rs delete mode 100644 tonic/src/transport/openssl.rs delete mode 100644 tonic/src/transport/service/tls/mod.rs create mode 100644 tonic/src/transport/tls/mod.rs create mode 100644 tonic/src/transport/tls/openssl.rs rename tonic/src/transport/{ => tls}/rustls.rs (100%) diff --git a/tonic-examples/src/helloworld/client.rs b/tonic-examples/src/helloworld/client.rs index a208f51..1200d47 100644 --- a/tonic-examples/src/helloworld/client.rs +++ b/tonic-examples/src/helloworld/client.rs @@ -8,9 +8,8 @@ pub mod hello_world { #[tokio::main] async fn main() -> Result<(), Box> { let origin = vec![ - http::Uri::from_static("http://[::1]:50051"), - http::Uri::from_static("http://[::1]:50051"), - http::Uri::from_static("http://[::1]:50051"), + http::Uri::from_static("http://[::1]:50051").into(), + ]; let svc = Channel::builder().balance_list(origin)?; diff --git a/tonic/Cargo.toml b/tonic/Cargo.toml index 370bdb4..0bbab3c 100644 --- a/tonic/Cargo.toml +++ b/tonic/Cargo.toml @@ -43,7 +43,7 @@ transport = [ "hyper", "tower", "tokio", - "openssl-1", + "native-tls", ] tower = [ "tower-reconnect", @@ -52,5 +52,5 @@ tower = [ "tower-load", "tower-discover" ] -openssl-1 = ["openssl", "tokio-openssl"] +native-tls = ["openssl", "tokio-openssl"] rustls = ["tokio-rustls"] diff --git a/tonic/src/transport/channel.rs b/tonic/src/transport/channel.rs index df600e4..e9a9b1b 100644 --- a/tonic/src/transport/channel.rs +++ b/tonic/src/transport/channel.rs @@ -1,15 +1,17 @@ -use super::service::{AddOrigin, BoxService, ServiceList}; +use super::{ + service::{BoxService, Connection, ServiceList}, + Endpoint, +}; use crate::{BoxBody, GrpcService}; use futures_util::try_future::{MapErr, TryFutureExt}; use http::Uri; -use hyper::client::conn; -use hyper::client::connect::HttpConnector; -use hyper::client::service::Connect; use hyper::{Request, Response}; use std::future::Future; use std::pin::Pin; use std::task::{Context, Poll}; +use tower_balance::p2c::Balance; use tower_buffer::{future::ResponseFuture, Buffer}; +use tower_discover::Discover; use tower_service::Service; type BoxFuture<'a, T> = Pin + Send + 'a>>; @@ -89,20 +91,26 @@ impl Builder { self } - pub fn balance_list(&mut self, list: Vec) -> Result { + pub fn balance_list(&mut self, list: Vec) -> Result { let discover = ServiceList::new(list); - let svc = tower_balance::p2c::Balance::from_entropy(discover); + self.balance(discover) + } + + fn balance(&mut self, discover: D) -> Result + where + D: Discover + Send + 'static, + D::Error: Into, + D::Key: Send + Clone, + { + let svc = Balance::from_entropy(discover); + let svc = BoxService::new(svc); let svc = Buffer::new(Box::new(svc) as Inner, 100); + Ok(Channel { svc }) } - // pub fn balance(&mut self, discover: D) -> &mut Self { - // self.balance = Some(discover); - // self - // } - - pub fn build(&self, uri: T) -> Result + pub fn build(&mut self, uri: T) -> Result where Uri: http::HttpTryFrom, { @@ -111,41 +119,6 @@ impl Builder { Err(e) => panic!("Invalid uri: {}", e.into()), }; - let settings = conn::Builder::new().http2_only(true).clone(); - - let svc = if let Some(ca) = &self.ca { - let domain = self - .override_domain - .clone() - .unwrap_or_else(|| uri.to_string()); - - #[cfg(not(any(feature = "openssl-1", feature = "rustls")))] - unreachable!("tls configured when no tls implementation feature was selected!"); - - #[cfg(feature = "openssl-1")] - let connector = super::openssl::TlsConnector::new(ca.clone(), domain)?; - - #[cfg(feature = "rustls")] - #[cfg(not(feature = "openssl-1"))] - let connector = super::openssl::TlsConnector::new(ca.clone(), domain)?; - - let maker = Connect::new(connector, settings); - let svc = tower_reconnect::Reconnect::new(maker, uri.clone()); - - let svc = AddOrigin::new(svc, uri); - let svc = BoxService::new(svc); - Buffer::new(Box::new(svc) as Inner, 100) - } else { - let connector = HttpConnector::new(); - let maker = Connect::new(connector, settings); - let svc = tower_reconnect::Reconnect::new(maker, uri.clone()); - - let svc = AddOrigin::new(svc, uri); - - let svc = BoxService::new(svc); - Buffer::new(Box::new(svc) as Inner, 100) - }; - - Ok(Channel { svc }) + self.balance_list(vec![uri.into()]) } } diff --git a/tonic/src/transport/endpoint.rs b/tonic/src/transport/endpoint.rs new file mode 100644 index 0000000..74bc0c9 --- /dev/null +++ b/tonic/src/transport/endpoint.rs @@ -0,0 +1,33 @@ +use super::tls::Cert; +use http::uri::Uri; + +#[derive(Debug, Clone)] +pub struct Endpoint { + uri: Uri, + cert: Option, +} + +impl Endpoint { + pub fn with_pem(uri: Uri, ca: Vec, domain: Option) -> Self { + let domain = domain.unwrap_or_else(|| uri.clone().to_string()); + + Self { + uri, + cert: Some(Cert { ca, domain }), + } + } + + pub(crate) fn uri(&self) -> &Uri { + &self.uri + } + + pub(crate) fn take_cert(&mut self) -> Option { + self.cert.take() + } +} + +impl From for Endpoint { + fn from(uri: Uri) -> Self { + Self { uri, cert: None } + } +} diff --git a/tonic/src/transport/mod.rs b/tonic/src/transport/mod.rs index c3c6ddc..13218d0 100644 --- a/tonic/src/transport/mod.rs +++ b/tonic/src/transport/mod.rs @@ -1,11 +1,10 @@ mod channel; -#[cfg(feature = "openssl-1")] -mod openssl; -#[cfg(feature = "rustls")] -mod rustls; +mod endpoint; mod service; +mod tls; pub use self::channel::Channel; +pub use self::endpoint::Endpoint; use std::{error, fmt}; diff --git a/tonic/src/transport/openssl.rs b/tonic/src/transport/openssl.rs deleted file mode 100644 index c8ece98..0000000 --- a/tonic/src/transport/openssl.rs +++ /dev/null @@ -1,72 +0,0 @@ -use http::Uri; -use hyper::client::connect::HttpConnector; -use openssl::ssl::{SslConnector, SslMethod}; -use openssl::x509::X509; -use std::{ - future::Future, - pin::Pin, - task::{Context, Poll}, -}; -use tokio::net::TcpStream; -use tokio_openssl::{connect, SslStream}; -use tower_make::MakeConnection; -use tower_service::Service; - -const ALPN_H2: &[u8] = b"\x02h2"; - -#[derive(Clone)] -pub struct TlsConnector { - http: HttpConnector, - config: SslConnector, - domain: String, -} - -impl TlsConnector { - pub fn new(ca: Vec, domain: String) -> Result { - let mut config = SslConnector::builder(SslMethod::tls()).unwrap(); - - config.set_alpn_protos(ALPN_H2).unwrap(); - - let ca = X509::from_pem(&ca[..]).unwrap(); - - config.cert_store_mut().add_cert(ca).unwrap(); - - let config = config.build(); - - let mut http = HttpConnector::new(); - http.enforce_http(false); - - Ok(Self { - http, - config, - domain, - }) - } -} - -impl Service for TlsConnector { - type Response = SslStream; - 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 config = self.config.configure().unwrap(); - let tcp = self.http.make_connection(uri.clone()); - let domain = self.domain.clone(); - - let fut = async move { - let io = tcp.await.unwrap(); - let tls = connect(config, &domain, io).await.unwrap(); - Ok(tls) - }; - - Box::pin(fut) - } -} diff --git a/tonic/src/transport/service/connect.rs b/tonic/src/transport/service/connect.rs index e6f4a5d..96c7a75 100644 --- a/tonic/src/transport/service/connect.rs +++ b/tonic/src/transport/service/connect.rs @@ -1,5 +1,5 @@ use super::{add_origin::AddOrigin, connector::Connector}; -use crate::body::BoxBody; +use crate::{transport::Endpoint, BoxBody}; use http::{Request, Response, Uri}; use hyper::client::conn::Builder; use hyper::client::service::Connect as HyperConnect; @@ -15,14 +15,15 @@ pub struct Connection { } impl Connection { - pub fn new(uri: Uri) -> Self { - let connector = Connector::new(); + pub fn new(mut endpoint: Endpoint) -> Result { + let connector = Connector::new(endpoint.take_cert())?; + let settings = Builder::new().http2_only(true).clone(); let connect = HyperConnect::new(connector, settings); - let reconnect = Reconnect::new(connect, uri.clone()); - let inner = AddOrigin::new(reconnect, uri); + let reconnect = Reconnect::new(connect, endpoint.uri().clone()); + let inner = AddOrigin::new(reconnect, endpoint.uri().clone()); - Self { inner } + Ok(Self { inner }) } } diff --git a/tonic/src/transport/service/connector.rs b/tonic/src/transport/service/connector.rs index 23f8c92..fd73cef 100644 --- a/tonic/src/transport/service/connector.rs +++ b/tonic/src/transport/service/connector.rs @@ -1,4 +1,5 @@ use super::io::BoxedIo; +use crate::transport::tls::{Cert, TlsAcceptor}; use http::Uri; use hyper::client::connect::HttpConnector; use std::future::Future; @@ -11,13 +12,21 @@ type ConnectFuture = >::Future; pub struct Connector { http: HttpConnector, + tls: Option, } impl Connector { - pub fn new() -> Self { - Self { - http: HttpConnector::new(), - } + pub fn new(cert: Option) -> Result { + let mut http = HttpConnector::new(); + http.enforce_http(false); + + let tls = if let Some(cert) = cert { + Some(TlsAcceptor::new(cert)?) + } else { + None + }; + + Ok(Self { http, tls }) } } @@ -33,16 +42,23 @@ impl Service for Connector { } fn call(&mut self, uri: Uri) -> Self::Future { - let connect_fut = MakeConnection::make_connection(&mut self.http, uri); + let io = MakeConnection::make_connection(&mut self.http, uri); + let tls = self.tls.clone(); - Box::pin(connect(connect_fut)) + Box::pin(connect(io, tls)) } } -async fn connect(connect: ConnectFuture) -> Result { +async fn connect( + connect: ConnectFuture, + tls: Option, +) -> Result { let io = connect.await?; - // TODO: build tls based on creds and features - - Ok(BoxedIo::new(io)) + if let Some(tls) = tls { + let conn = tls.connect(io).await?; + Ok(BoxedIo::new(conn)) + } else { + Ok(BoxedIo::new(io)) + } } diff --git a/tonic/src/transport/service/discover.rs b/tonic/src/transport/service/discover.rs index 083a2ed..9756096 100644 --- a/tonic/src/transport/service/discover.rs +++ b/tonic/src/transport/service/discover.rs @@ -1,17 +1,17 @@ use super::connect::Connection; -use http::Uri; +use crate::transport::Endpoint; use std::collections::VecDeque; use std::task::{Context, Poll}; use tower_discover::{Change, Discover}; #[derive(Debug)] pub struct ServiceList { - list: VecDeque, + list: VecDeque, i: usize, } impl ServiceList { - pub fn new(list: Vec) -> Self { + pub fn new(list: Vec) -> Self { Self { list: list.into(), i: 0, @@ -22,18 +22,21 @@ impl ServiceList { impl Discover for ServiceList { type Key = usize; type Service = Connection; - type Error = hyper::Error; + type Error = crate::Error; fn poll( &mut self, _cx: &mut Context<'_>, ) -> Poll, Self::Error>> { match self.list.pop_front() { - Some(uri) => { + Some(endpoint) => { let i = self.i; self.i += 1; - let service = Connection::new(uri); - Poll::Ready(Ok(Change::Insert(i, service))) + + match Connection::new(endpoint) { + Ok(svc) => Poll::Ready(Ok(Change::Insert(i, svc))), + Err(e) => Poll::Ready(Err(e)), + } } None => Poll::Pending, } diff --git a/tonic/src/transport/service/mod.rs b/tonic/src/transport/service/mod.rs index be08e4e..f75738d 100644 --- a/tonic/src/transport/service/mod.rs +++ b/tonic/src/transport/service/mod.rs @@ -4,8 +4,8 @@ mod connect; mod connector; mod discover; mod io; -mod tls; pub use self::add_origin::AddOrigin; pub use self::boxed::BoxService; +pub use self::connect::Connection; pub use self::discover::ServiceList; diff --git a/tonic/src/transport/service/tls/mod.rs b/tonic/src/transport/service/tls/mod.rs deleted file mode 100644 index 8b13789..0000000 --- a/tonic/src/transport/service/tls/mod.rs +++ /dev/null @@ -1 +0,0 @@ - diff --git a/tonic/src/transport/tls/mod.rs b/tonic/src/transport/tls/mod.rs new file mode 100644 index 0000000..b47a9ca --- /dev/null +++ b/tonic/src/transport/tls/mod.rs @@ -0,0 +1,33 @@ +// #[cfg(feature = "openssl-1")] +// #[cfg(not(feature = "rustls"))] +// #[path = "rustls.rs"] +// mod imp; + +#[cfg(feature = "native-tls")] +#[cfg(not(feature = "rustls"))] +#[path = "openssl.rs"] +mod imp; + +use tokio::net::TcpStream; + +#[derive(Debug, Clone)] +pub struct Cert { + pub(crate) ca: Vec, + pub(crate) domain: String, +} + +#[derive(Clone)] +pub struct TlsAcceptor { + inner: imp::TlsAcceptor, +} + +impl TlsAcceptor { + pub fn new(cert: Cert) -> Result { + let inner = imp::TlsAcceptor::new(cert)?; + Ok(Self { inner }) + } + + pub async fn connect(&self, io: TcpStream) -> Result { + self.inner.connect(io).await + } +} diff --git a/tonic/src/transport/tls/openssl.rs b/tonic/src/transport/tls/openssl.rs new file mode 100644 index 0000000..17bfc1f --- /dev/null +++ b/tonic/src/transport/tls/openssl.rs @@ -0,0 +1,42 @@ +use super::Cert; +use openssl::ssl::{SslConnector, SslMethod}; +use openssl::x509::X509; +use std::sync::Arc; +use tokio::net::TcpStream; +use tokio_openssl::SslStream; + +const ALPN_H2: &[u8] = b"\x02h2"; + +pub type TlsStream = SslStream; + +#[derive(Clone)] +pub struct TlsAcceptor { + config: SslConnector, + domain: Arc, +} + +impl TlsAcceptor { + pub fn new(cert: Cert) -> Result { + let Cert { ca, domain } = cert; + let mut config = SslConnector::builder(SslMethod::tls()).unwrap(); + + config.set_alpn_protos(ALPN_H2)?; + + let ca = X509::from_pem(&ca[..])?; + + config.cert_store_mut().add_cert(ca)?; + + let config = config.build(); + + Ok(Self { + config, + domain: Arc::new(domain), + }) + } + + pub async fn connect(&self, io: TcpStream) -> Result { + let config = self.config.configure()?; + let tls = tokio_openssl::connect(config, &self.domain, io).await?; + Ok(tls) + } +} diff --git a/tonic/src/transport/rustls.rs b/tonic/src/transport/tls/rustls.rs similarity index 100% rename from tonic/src/transport/rustls.rs rename to tonic/src/transport/tls/rustls.rs