meshtls: Move TLS e2e tests into the meshtls crate (#1366)

We'll want to run the same end-to-end tests against all meshtls
backends. This change moves the rustls `tls_accept` tests into meshtls
crate.

Author: olix0r

Cargo.lock

@@ -1022,34 +1022,36 @@ name = "linkerd-meshtls"
 version = "0.1.0"
 dependencies = [
  "futures",
+ "linkerd-conditional",
  "linkerd-error",
  "linkerd-identity",
  "linkerd-io",
  "linkerd-meshtls-rustls",
+ "linkerd-proxy-transport",
  "linkerd-stack",
  "linkerd-tls",
+ "linkerd-tls-test-util",
+ "linkerd-tracing",
  "pin-project",
+ "tokio",
+ "tracing",
 ]
 
 [[package]]
 name = "linkerd-meshtls-rustls"
 version = "0.1.0"
 dependencies = [
  "futures",
- "linkerd-conditional",
  "linkerd-error",
  "linkerd-identity",
  "linkerd-io",
- "linkerd-proxy-transport",
  "linkerd-stack",
  "linkerd-tls",
  "linkerd-tls-test-util",
- "linkerd-tracing",
  "ring",
  "thiserror",
  "tokio",
  "tokio-rustls",
- "tower",
  "tracing",
  "webpki",
 ]

linkerd/meshtls/Cargo.toml

@@ -20,3 +20,11 @@ linkerd-meshtls-rustls = { path = "rustls", optional = true }
 linkerd-stack = { path = "../stack" }
 linkerd-tls = { path = "../tls" }
 pin-project = "1"
+
+[dev-dependencies]
+linkerd-conditional = { path = "../conditional" }
+linkerd-proxy-transport = { path = "../proxy/transport" }
+linkerd-tls-test-util = { path = "../tls/test-util" }
+linkerd-tracing = { path = "../tracing", features = ["ansi"] }
+tokio = { version = "1", features = ["macros", "net", "rt-multi-thread"] }
+tracing = "0.1"

linkerd/meshtls/rustls/Cargo.toml

@@ -25,9 +25,4 @@ tracing = "0.1"
 webpki = "0.21"
 
 [dev-dependencies]
-linkerd-conditional = { path = "../../conditional" }
-linkerd-proxy-transport = { path = "../../proxy/transport" }
 linkerd-tls-test-util = { path = "../../tls/test-util" }
-linkerd-tracing = { path = "../../tracing", features = ["ansi"] }
-tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
-tower = { version = "0.4.10", default-features = false, features = ["util"] }

linkerd/meshtls/tests/rustls.rs

@@ -0,0 +1,22 @@
+#![cfg(feature = "rustls")]
+#![deny(warnings, rust_2018_idioms)]
+#![forbid(unsafe_code)]
+
+mod util;
+
+use linkerd_meshtls::Mode;
+
+#[tokio::test(flavor = "current_thread")]
+async fn plaintext() {
+    util::plaintext(Mode::Rustls).await;
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn proxy_to_proxy_tls_works() {
+    util::proxy_to_proxy_tls_works(Mode::Rustls).await;
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn proxy_to_proxy_tls_pass_through_when_identity_does_not_match() {
+    util::proxy_to_proxy_tls_pass_through_when_identity_does_not_match(Mode::Rustls).await;
+}

linkerd/meshtls/rustls/tests/tls_accept.rs renamed to linkerd/meshtls/tests/util.rs

@@ -1,59 +1,29 @@
-#![cfg(test)]
-
-// These are basically integration tests for the `connection` submodule, but
-// they cannot be "real" integration tests because `connection` isn't a public
-// interface and because `connection` exposes a `#[cfg(test)]`-only API for use
-// by these tests.
+#![deny(warnings)]
+#![forbid(unsafe_code)]
 
 use futures::prelude::*;
 use linkerd_conditional::Conditional;
 use linkerd_error::Infallible;
 use linkerd_identity::{Credentials, DerX509, Name};
 use linkerd_io::{self as io, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
-use linkerd_meshtls_rustls as meshtls;
+use linkerd_meshtls as meshtls;
 use linkerd_proxy_transport::{
     addrs::*,
     listen::{Addrs, Bind, BindTcp},
     ConnectTcp, Keepalive, ListenAddr,
 };
-use linkerd_stack::{ExtractParam, InsertParam, NewService, Param};
+use linkerd_stack::{
+    layer::Layer, service_fn, ExtractParam, InsertParam, NewService, Param, ServiceExt,
+};
 use linkerd_tls as tls;
 use linkerd_tls_test_util as test_util;
 use std::{future::Future, net::SocketAddr, sync::mpsc, time::Duration};
 use tokio::net::TcpStream;
-use tower::{
-    layer::Layer,
-    util::{service_fn, ServiceExt},
-};
-use tracing::instrument::Instrument;
-
-type ServerConn<T, I> = (
-    (tls::ConditionalServerTls, T),
-    io::EitherIo<meshtls::ServerIo<tls::server::DetectIo<I>>, tls::server::DetectIo<I>>,
-);
-
-fn load(ent: &test_util::Entity) -> (meshtls::creds::Store, meshtls::NewClient, meshtls::Server) {
-    let roots_pem = std::str::from_utf8(ent.trust_anchors).expect("valid PEM");
-    let (mut store, rx) = meshtls::creds::watch(
-        ent.name.parse().unwrap(),
-        roots_pem,
-        ent.key,
-        b"fake CSR data",
-    )
-    .expect("credentials must be readable");
+use tracing::Instrument;
 
-    let expiry = std::time::SystemTime::now() + Duration::from_secs(600);
-    store
-        .set_certificate(DerX509(ent.crt.to_vec()), vec![], expiry)
-        .expect("certificate must be valid");
-
-    (store, rx.new_client(), rx.server())
-}
-
-#[tokio::test(flavor = "current_thread")]
-async fn plaintext() {
-    let (_foo, _, server_tls) = load(&test_util::FOO_NS1);
-    let (_bar, client_tls, _) = load(&test_util::BAR_NS1);
+pub async fn plaintext(mode: meshtls::Mode) {
+    let (_foo, _, server_tls) = load(mode, &test_util::FOO_NS1);
+    let (_bar, client_tls, _) = load(mode, &test_util::BAR_NS1);
     let (client_result, server_result) = run_test(
         client_tls,
         Conditional::None(tls::NoClientTls::NotProvidedByServiceDiscovery),
@@ -76,10 +46,9 @@ async fn plaintext() {
     assert_eq!(&server_result.result.expect("ping")[..], PING);
 }
 
-#[tokio::test(flavor = "current_thread")]
-async fn proxy_to_proxy_tls_works() {
-    let (_foo, _, server_tls) = load(&test_util::FOO_NS1);
-    let (_bar, client_tls, _) = load(&test_util::BAR_NS1);
+pub async fn proxy_to_proxy_tls_works(mode: meshtls::Mode) {
+    let (_foo, _, server_tls) = load(mode, &test_util::FOO_NS1);
+    let (_bar, client_tls, _) = load(mode, &test_util::BAR_NS1);
     let server_id = tls::ServerId(test_util::FOO_NS1.name.parse().unwrap());
     let (client_result, server_result) = run_test(
         client_tls.clone(),
@@ -107,13 +76,12 @@ async fn proxy_to_proxy_tls_works() {
     assert_eq!(&server_result.result.expect("ping")[..], PING);
 }
 
-#[tokio::test(flavor = "current_thread")]
-async fn proxy_to_proxy_tls_pass_through_when_identity_does_not_match() {
-    let (_foo, _, server_tls) = load(&test_util::FOO_NS1);
+pub async fn proxy_to_proxy_tls_pass_through_when_identity_does_not_match(mode: meshtls::Mode) {
+    let (_foo, _, server_tls) = load(mode, &test_util::FOO_NS1);
 
     // Misuse the client's identity instead of the server's identity. Any
     // identity other than `server_tls.server_identity` would work.
-    let (_bar, client_tls, _) = load(&test_util::BAR_NS1);
+    let (_bar, client_tls, _) = load(mode, &test_util::BAR_NS1);
     let sni = test_util::BAR_NS1.name.parse::<Name>().unwrap();
 
     let (client_result, server_result) = run_test(
@@ -138,6 +106,33 @@ async fn proxy_to_proxy_tls_pass_through_when_identity_does_not_match() {
     assert_eq!(&server_result.result.unwrap()[..], START_OF_TLS);
 }
 
+type ServerConn<T, I> = (
+    (tls::ConditionalServerTls, T),
+    io::EitherIo<meshtls::ServerIo<tls::server::DetectIo<I>>, tls::server::DetectIo<I>>,
+);
+
+fn load(
+    mode: meshtls::Mode,
+    ent: &test_util::Entity,
+) -> (meshtls::creds::Store, meshtls::NewClient, meshtls::Server) {
+    let roots_pem = std::str::from_utf8(ent.trust_anchors).expect("valid PEM");
+    let (mut store, rx) = mode
+        .watch(
+            ent.name.parse().unwrap(),
+            roots_pem,
+            ent.key,
+            b"fake CSR data",
+        )
+        .expect("credentials must be readable");
+
+    let expiry = std::time::SystemTime::now() + Duration::from_secs(600);
+    store
+        .set_certificate(DerX509(ent.crt.to_vec()), vec![], expiry)
+        .expect("certificate must be valid");
+
+    (store, rx.new_client(), rx.server())
+}
+
 struct Transported<I, R> {
     tls: Option<I>,
 
@@ -150,7 +145,8 @@ struct ServerParams {
     identity: meshtls::Server,
 }
 
-type ClientIo = io::EitherIo<io::ScopedIo<TcpStream>, meshtls::ClientIo<io::ScopedIo<TcpStream>>>;
+type ClientIo =
+    io::EitherIo<io::ScopedIo<TcpStream>, linkerd_meshtls::ClientIo<io::ScopedIo<TcpStream>>>;
 
 /// Runs a test for a single TCP connection. `client` processes the connection
 /// on the client side and `server` processes the connection on the server
@@ -159,7 +155,7 @@ async fn run_test<C, CF, CR, S, SF, SR>(
     client_tls: meshtls::NewClient,
     client_server_id: Conditional<tls::ServerId, tls::NoClientTls>,
     client: C,
-    server_id: meshtls::Server,
+    server_tls: meshtls::Server,
     server: S,
 ) -> (
     Transported<tls::ConditionalClientTls, CR>,
@@ -184,7 +180,7 @@ where
 
         let detect = tls::NewDetectTls::<meshtls::Server, _, _>::new(
             ServerParams {
-                identity: server_id,
+                identity: server_tls,
             },
             move |meta: (tls::ConditionalServerTls, Addrs)| {
                 let server = server.clone();
@@ -233,7 +229,6 @@ where
         // type, e.g. `Arc<Mutex>`, but using a channel simplifies the code and
         // parallels the server side.
         let (sender, receiver) = mpsc::channel::<Transported<tls::ConditionalClientTls, CR>>();
-        let sender_clone = sender.clone();
 
         let tls = Some(client_server_id.clone().map(Into::into));
         let client = async move {
@@ -243,7 +238,7 @@ where
                 .await;
             match conn {
                 Err(e) => {
-                    sender_clone
+                    sender
                         .send(Transported {
                             tls: None,
                             result: Err(e),