Ftr: add tls for client

.licenserc.yaml

@@ -65,6 +65,7 @@ header: # `header` section is configurations for source codes license header.
     - '.github'
     - "**/*.yaml"
     - "**/generated/**"
+    - "**/fixtures/**"
   comment: on-failure # on what condition license-eye will comment on the pull request, `on-failure`, `always`, `never`.
 
   # license-location-threshold specifies the index threshold where the license header can be located,

dubbo/Cargo.toml

@@ -19,7 +19,9 @@ futures-util = "0.3.23"
 futures-core ="0.3.23"
 argh = "0.1"
 rustls-pemfile = "1.0.0"
-tokio-rustls="0.23.4"
+rustls-webpki = "0.101.3"
+rustls-native-certs = "0.6.3"
+tokio-rustls="0.24.1"
 tokio = { version = "1.0", features = [ "rt-multi-thread", "time", "fs", "macros", "net", "signal",  "full" ] }
 prost = "0.10.4"
 async-trait = "0.1.56"

dubbo/src/cluster/directory.rs

@@ -62,7 +62,8 @@ impl StaticDirectory {
 impl Directory for StaticDirectory {
     fn list(&self, invocation: Arc<RpcInvocation>) -> Vec<BoxInvoker> {
         let url = Url::from_url(&format!(
-            "tri://{}:{}/{}",
+            "{}://{}:{}/{}",
+            self.uri.scheme_str().unwrap_or("tri"),
             self.uri.host().unwrap(),
             self.uri.port().unwrap(),
             invocation.get_target_service_unique_name(),

dubbo/src/protocol/triple/triple_invoker.rs

@@ -38,9 +38,13 @@ pub struct TripleInvoker {
 impl TripleInvoker {
     pub fn new(url: Url) -> TripleInvoker {
         let uri = http::Uri::from_str(&url.to_url()).unwrap();
+        let mut conn = Connection::new().with_host(uri.clone());
+        if let Some(scheme) = uri.scheme_str() {
+            conn = conn.with_connector(scheme.to_string());
+        }
         Self {
             url,
-            conn: BoxCloneService::new(Connection::new().with_host(uri)),
+            conn: BoxCloneService::new(conn),
         }
     }
 }

dubbo/src/triple/client/builder.rs

@@ -59,7 +59,7 @@ impl ClientBuilder {
             connector: "",
             directory: Some(Box::new(StaticDirectory::new(&host))),
             direct: true,
-            host: host.clone().to_string(),
+            host: host.to_string(),
         }
     }
 

dubbo/src/triple/transport/connection.rs

@@ -26,7 +26,7 @@ use crate::{boxed, triple::transport::connector::get_connector};
 #[derive(Debug, Clone)]
 pub struct Connection {
     host: hyper::Uri,
-    connector: &'static str,
+    connector: String,
     builder: Builder,
 }
 
@@ -40,12 +40,12 @@ impl Connection {
     pub fn new() -> Self {
         Connection {
             host: hyper::Uri::default(),
-            connector: "http",
+            connector: "http".to_string(),
             builder: Builder::new(),
         }
     }
 
-    pub fn with_connector(mut self, connector: &'static str) -> Self {
+    pub fn with_connector(mut self, connector: String) -> Self {
         self.connector = connector;
         self
     }
@@ -82,7 +82,7 @@ where
 
     fn call(&mut self, req: http::Request<ReqBody>) -> Self::Future {
         let builder = self.builder.clone().http2_only(true).to_owned();
-        let mut connector = Connect::new(get_connector(self.connector), builder);
+        let mut connector = Connect::new(get_connector(self.connector.as_str()), builder);
         let uri = self.host.clone();
         let fut = async move {
             debug!("send base call to {}", uri);

dubbo/src/triple/transport/connector/https_connector.rs

@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+use std::{
+    net::{Ipv4Addr, SocketAddr, SocketAddrV4},
+    str::FromStr,
+    sync::Arc,
+};
+
+use dubbo_logger::tracing;
+use http::Uri;
+use hyper::client::connect::dns::Name;
+use rustls_native_certs::load_native_certs;
+use tokio::net::TcpStream;
+use tokio_rustls::{
+    client::TlsStream,
+    rustls::{self},
+    TlsConnector as TlsConnectorTokio,
+};
+use tower_service::Service;
+
+use crate::triple::transport::resolver::{dns::DnsResolver, Resolve};
+
+#[derive(Clone, Default)]
+pub struct HttpsConnector<R = DnsResolver> {
+    resolver: R,
+}
+
+impl HttpsConnector {
+    pub fn new() -> Self {
+        Self {
+            resolver: DnsResolver::default(),
+        }
+    }
+}
+
+impl<R> HttpsConnector<R> {
+    pub fn new_with_resolver(resolver: R) -> HttpsConnector<R> {
+        Self { resolver }
+    }
+}
+
+impl<R> Service<Uri> for HttpsConnector<R>
+where
+    R: Resolve + Clone + Send + Sync + 'static,
+    R::Future: Send,
+{
+    type Response = TlsStream<TcpStream>;
+
+    type Error = crate::Error;
+
+    type Future = crate::BoxFuture<Self::Response, Self::Error>;
+
+    fn poll_ready(
+        &mut self,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), Self::Error>> {
+        self.resolver.poll_ready(cx).map_err(|err| err.into())
+    }
+
+    fn call(&mut self, uri: Uri) -> Self::Future {
+        let mut inner = self.clone();
+
+        Box::pin(async move { inner.call_async(uri).await })
+    }
+}
+
+impl<R> HttpsConnector<R>
+where
+    R: Resolve + Send + Sync + 'static,
+{
+    async fn call_async(&mut self, uri: Uri) -> Result<TlsStream<TcpStream>, crate::Error> {
+        let host = uri.host().unwrap();
+        let port = uri.port_u16().unwrap();
+
+        let addr = if let Ok(addr) = host.parse::<Ipv4Addr>() {
+            tracing::info!("host is ip address: {:?}", host);
+            SocketAddr::V4(SocketAddrV4::new(addr, port))
+        } else {
+            tracing::info!("host is dns: {:?}", host);
+            let addrs = self
+                .resolver
+                .resolve(Name::from_str(host).unwrap())
+                .await
+                .map_err(|err| err.into())?;
+            let addrs: Vec<SocketAddr> = addrs
+                .map(|mut addr| {
+                    addr.set_port(port);
+                    addr
+                })
+                .collect();
+            addrs[0]
+        };
+
+        let mut root_store = rustls::RootCertStore::empty();
+
+        for cert in load_native_certs()? {
+            root_store.add(&rustls::Certificate(cert.0))?;
+        }
+
+        let config = rustls::ClientConfig::builder()
+            .with_safe_defaults()
+            .with_root_certificates(root_store)
+            .with_no_client_auth();
+
+        let connector = TlsConnectorTokio::from(Arc::new(config));
+
+        let stream = TcpStream::connect(&addr).await?;
+        let domain = rustls::ServerName::try_from(host).map_err(|err| {
+            crate::status::Status::new(crate::status::Code::Internal, err.to_string())
+        })?;
+        let stream = connector.connect(domain, stream).await?;
+
+        Ok(stream)
+    }
+}

dubbo/src/triple/transport/connector/mod.rs

@@ -16,6 +16,7 @@
  */
 
 pub mod http_connector;
+pub mod https_connector;
 #[cfg(any(target_os = "macos", target_os = "unix"))]
 pub mod unix_connector;
 
@@ -73,12 +74,16 @@ where
     }
 }
 
-pub fn get_connector(connector: &'static str) -> BoxCloneService<Uri, BoxIO, crate::Error> {
+pub fn get_connector(connector: &str) -> BoxCloneService<Uri, BoxIO, crate::Error> {
     match connector {
         "http" => {
             let c = http_connector::HttpConnector::new();
             BoxCloneService::new(Connector::new(c))
         }
+        "https" => {
+            let c = https_connector::HttpsConnector::new();
+            BoxCloneService::new(Connector::new(c))
+        }
         #[cfg(any(target_os = "macos", target_os = "unix"))]
         "unix" => {
             let c = unix_connector::UnixConnector::new();

dubbo/src/utils/tls.rs

@@ -15,10 +15,10 @@
  * limitations under the License.
  */
 
-use rustls_pemfile::{certs, rsa_private_keys};
+use rustls_pemfile::{certs, ec_private_keys, pkcs8_private_keys, rsa_private_keys};
 use std::{
     fs::File,
-    io::{self, BufReader},
+    io::{self, BufReader, Cursor, Read},
     path::Path,
 };
 use tokio_rustls::rustls::{Certificate, PrivateKey};
@@ -30,7 +30,22 @@ pub fn load_certs(path: &Path) -> io::Result<Vec<Certificate>> {
 }
 
 pub fn load_keys(path: &Path) -> io::Result<Vec<PrivateKey>> {
-    rsa_private_keys(&mut BufReader::new(File::open(path)?))
-        .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "invalid key"))
-        .map(|mut keys| keys.drain(..).map(PrivateKey).collect())
+    let file = &mut BufReader::new(File::open(path)?);
+    let mut data = Vec::new();
+    file.read_to_end(&mut data)?;
+
+    let mut cursor = Cursor::new(data);
+
+    let parsers = [rsa_private_keys, pkcs8_private_keys, ec_private_keys];
+
+    for parser in &parsers {
+        if let Ok(mut key) = parser(&mut cursor) {
+            if !key.is_empty() {
+                return Ok(key.drain(..).map(PrivateKey).collect());
+            }
+        }
+        cursor.set_position(0);
+    }
+
+    Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid key"))
 }

examples/echo/Cargo.toml

@@ -19,6 +19,14 @@ path = "src/echo/server.rs"
 name = "echo-client"
 path = "src/echo/client.rs"
 
+[[bin]]
+name = "echo-tls-server"
+path = "src/echo-tls/server.rs"
+
+[[bin]]
+name = "echo-tls-client"
+path = "src/echo-tls/client.rs"
+
 [dependencies]
 http = "0.2"
 http-body = "0.4.4"

examples/echo/README.md

@@ -15,3 +15,21 @@ reply: EchoResponse { message: "msg1 from server" }
 reply: EchoResponse { message: "msg2 from server" }
 reply: EchoResponse { message: "msg3 from server" }
 ```
+
+## build and run `echo-tls`
+
+**Please first install the `ca.crt` certificate file under the `fixtures` path to the platform's native certificate store.**
+
+```sh
+$ cd github.com/apache/dubbo-rust/examples/echo-tls/
+$ cargo build
+
+$ # run sever
+$ ../../target/debug/echo-tls-server
+
+$ # run client
+$ ../../target/debug/echo-tls-client
+reply: EchoResponse { message: "msg1 from tls-server" }
+reply: EchoResponse { message: "msg2 from tls-server" }
+reply: EchoResponse { message: "msg3 from tls-server" }
+```

examples/echo/README_CN.md

@@ -15,3 +15,21 @@ reply: EchoResponse { message: "msg1 from server" }
 reply: EchoResponse { message: "msg2 from server" }
 reply: EchoResponse { message: "msg3 from server" }
 ```
+
+## 构建并运行`echo-tls`
+
+**请先将`fixtures`路径下的`ca.crt`证书文件安装到系统信任根证书中.**
+
+```sh
+$ cd github.com/apache/dubbo-rust/examples/echo-tls/
+$ cargo build
+
+$ # 运行服务端
+$ ../../target/debug/echo-tls-server
+
+$ # 运行客户端
+$ ../../target/debug/echo-tls-client
+reply: EchoResponse { message: "msg1 from tls-server" }
+reply: EchoResponse { message: "msg2 from tls-server" }
+reply: EchoResponse { message: "msg3 from tls-server" }
+```

examples/echo/fixtures/ca.crt

@@ -0,0 +1,21 @@
+-----BEGIN CERTIFICATE-----
+MIIDiTCCAnGgAwIBAgIUGJWxrMGe9qRZzAfd5w4XIT3lkcEwDQYJKoZIhvcNAQEL
+BQAwVDEVMBMGA1UEAwwMVGVzdCBSb290IENBMQswCQYDVQQGEwJVUzENMAsGA1UE
+CAwEVGVzdDENMAsGA1UEBwwEVGVzdDEQMA4GA1UECgwHT3BlbmRhbDAeFw0yMzA4
+MTQxMTEzMzRaFw0yNDA4MTMxMTEzMzRaMFQxFTATBgNVBAMMDFRlc3QgUm9vdCBD
+QTELMAkGA1UEBhMCVVMxDTALBgNVBAgMBFRlc3QxDTALBgNVBAcMBFRlc3QxEDAO
+BgNVBAoMB09wZW5kYWwwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQCt
+UF6mcDgSyH5t78XnJusvQxsUfv2XydHtvLcIpwkCkIuIj7nF2WH064Gv12x+y42W
+mb+5z6JTgHRMRqcyQM8q4PQFrKvxPX8R2Limd7VLBJzYjR7Ma7JIrDohLnfywxUP
+19P5SzaGiro+ZK3t3xCnmtHcYoM+An0mQdKyVV7ytzAfg1PqkfDme19I28fH8cOP
+tF+RU8/LEHnte519O1bawx7xNdPsyykMrFij02o1VUeum2K9Wya8xHDixokveYDW
+swg5G4Tsy1QfgqFgxAXahIroPIwQvZOGkWVsmPXRXHtHNFG91ntJivv2HBFniUTq
+A0UbVdj09T+h+JLc19G9AgMBAAGjUzBRMB0GA1UdDgQWBBQ2672x8uh6Lud0EkjO
+wt2aEioeKjAfBgNVHSMEGDAWgBQ2672x8uh6Lud0EkjOwt2aEioeKjAPBgNVHRMB
+Af8EBTADAQH/MA0GCSqGSIb3DQEBCwUAA4IBAQCQkLp3GzZOXXXOKiMF6Iev1OUW
+w1jr7hVdJHOVGNCD6uZLuwSXJOWnEP8+hp8WvMl7SQAPpVYsTjdqhLATLaAZDucG
+sDq6oUTh/v8QVIBm0qF8+iMU8XZfgoeKuY13RXs23hneMAPQ5rcPwQhQEQkkqUvi
+Fq8qYFVd5mEr6Z62DT0s544WaBrpHr37mHOv0hIkHtX7Dy2Juc23MYw+W4PSD4fm
+sr1kARwHtY1meX+H3iRsX+7juTa33v+7H4IivhcPobIxFp+Hs9R5mx5u80wKMjVv
+t3STmB4nE7pABzucrjkSo43jIUwYN4rwydlSma9VkzvY6ry86HQuemycRb9H
+-----END CERTIFICATE-----

examples/echo/fixtures/server.crt

@@ -0,0 +1,23 @@
+-----BEGIN CERTIFICATE-----
+MIID1zCCAr+gAwIBAgIUJKTfvASV+RnwF7oLO84HZJDyvLwwDQYJKoZIhvcNAQEL
+BQAwVDEVMBMGA1UEAwwMVGVzdCBSb290IENBMQswCQYDVQQGEwJVUzENMAsGA1UE
+CAwEVGVzdDENMAsGA1UEBwwEVGVzdDEQMA4GA1UECgwHT3BlbmRhbDAeFw0yMzA4
+MTQxNDU5MzZaFw0yNDA2MDkxNDU5MzZaMFkxGjAYBgNVBAMMEVJlZGlzIGNlcnRp
+ZmljYXRlMQswCQYDVQQGEwJVUzENMAsGA1UECAwEVGVzdDENMAsGA1UEBwwEVGVz
+dDEQMA4GA1UECgwHT3BlbmRhbDCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoC
+ggEBAIwREKDrRgZ2jlR3tpLHvMiW8JDu4JiLBxyrlJJE5ndhuH7MEgwz8HnXvxbD
+eyuamzkAzQIvqfVFVTRuVEYyEtoGzIegDL76H9ybuMGhKBK1m0TmiH7bOsAVMqZN
+vDtQJiw8qePtSq3G3H7Sw+/oudrJIc/f7kDox/lndKHTBmLbjSrvpkOJk2qnvhPJ
+ih4SuLNiW+tHv4sUdYBXXxn2wLHXNLGrlpeW28jtWGfu2noRCzikOYL/jwg2xzXV
+cBSuFwQ3swLDG/htqpePVA/sLxbXTt03A8fCajYcKiJdW88gqw4dW01ya8rCr5MU
+1C7lPwNCB8qNn8pdkmrh/Oc0zDsCAwEAAaOBmzCBmDAfBgNVHSMEGDAWgBQ2672x
+8uh6Lud0EkjOwt2aEioeKjAJBgNVHRMEAjAAMAsGA1UdDwQEAwIE8DA+BgNVHREE
+NzA1gglsb2NhbGhvc3SHBH8AAAGHBKweAAKHBKweAAOHBKweAASHBKweAAWHBKwe
+AAaHBKweAAcwHQYDVR0OBBYEFGvNF07RBwyi3tbpFIJtvWhXAGblMA0GCSqGSIb3
+DQEBCwUAA4IBAQAd57+0YXfg8eIe2UkqLshIEonoIpKhmsIpRJyXLOUWYaSHri4w
+aDqPjogA39w34UcZsumfSReWBGrCyBroSCqQZOM166tw79+AVdjHHtgNm8pFRhO7
+0vnFdAU30TOQP+mRF3mXz3hcK68U/4cRhXC5jXq8YRLiAG74G3PmXmmk2phtluEL
+SLLCvF5pCz3EaYsEKP+ZQpdY3BLp6Me7XDpGWPuNYVwVTJwwM9CLjQ8pxMlz1O1x
+HVN7xGtLz4dw9nEqnmjYBvH8aum+iAQPiHVuGfQfqIea28XeuyV4c5TL2b+OUsLY
+BRhX+z5OkGHXcMc1QDKo3PZcs8C1w8SC1x9D
+-----END CERTIFICATE-----