Merge pull request redis-rs#548 from particle-iot/cluster-tls

Adds support for TLS with Cluster mode

Author: Marwes

src/cluster.rs

@@ -71,6 +71,23 @@ pub struct ClusterConnection {
     password: Option<String>,
     read_timeout: RefCell<Option<Duration>>,
     write_timeout: RefCell<Option<Duration>>,
+    tls: Option<TlsMode>,
+}
+
+#[derive(Clone, Copy)]
+enum TlsMode {
+    Secure,
+    Insecure,
+}
+
+impl TlsMode {
+    fn from_insecure_flag(insecure: bool) -> TlsMode {
+        if insecure {
+            TlsMode::Insecure
+        } else {
+            TlsMode::Secure
+        }
+    }
 }
 
 impl ClusterConnection {
@@ -83,14 +100,33 @@ impl ClusterConnection {
             Self::create_initial_connections(&initial_nodes, readonly, password.clone())?;
 
         let connection = ClusterConnection {
-            initial_nodes,
             connections: RefCell::new(connections),
             slots: RefCell::new(SlotMap::new()),
             auto_reconnect: RefCell::new(true),
             readonly,
             password,
             read_timeout: RefCell::new(None),
             write_timeout: RefCell::new(None),
+            #[cfg(feature = "tls")]
+            tls: {
+                if initial_nodes.is_empty() {
+                    None
+                } else {
+                    // TODO: Maybe should run through whole list and make sure they're all matching?
+                    match &initial_nodes.get(0).unwrap().addr {
+                        ConnectionAddr::Tcp(_, _) => None,
+                        ConnectionAddr::TcpTls {
+                            host: _,
+                            port: _,
+                            insecure,
+                        } => Some(TlsMode::from_insecure_flag(*insecure)),
+                        _ => None,
+                    }
+                }
+            },
+            #[cfg(not(feature = "tls"))]
+            tls: None,
+            initial_nodes,
         };
         connection.refresh_slots()?;
 
@@ -166,6 +202,14 @@ impl ClusterConnection {
         for info in initial_nodes.iter() {
             let addr = match info.addr {
                 ConnectionAddr::Tcp(ref host, port) => format!("redis://{}:{}", host, port),
+                ConnectionAddr::TcpTls {
+                    ref host,
+                    port,
+                    insecure,
+                } => {
+                    let tls_mode = TlsMode::from_insecure_flag(insecure);
+                    build_connection_string(host, Some(port), Some(tls_mode))
+                }
                 _ => panic!("No reach."),
             };
 
@@ -180,7 +224,7 @@ impl ClusterConnection {
         if connections.is_empty() {
             return Err(RedisError::from((
                 ErrorKind::IoError,
-                "It is failed to check startup nodes.",
+                "It failed to check startup nodes.",
             )));
         }
         Ok(connections)
@@ -246,7 +290,7 @@ impl ClusterConnection {
         let mut samples = connections.values_mut().choose_multiple(&mut rng, len);
 
         for mut conn in samples.iter_mut() {
-            if let Ok(mut slots_data) = get_slots(&mut conn) {
+            if let Ok(mut slots_data) = get_slots(&mut conn, self.tls) {
                 slots_data.sort_by_key(|s| s.start());
                 let last_slot = slots_data.iter().try_fold(0, |prev_end, slot_data| {
                     if prev_end != slot_data.start() {
@@ -395,15 +439,19 @@ impl ClusterConnection {
                         let kind = err.kind();
 
                         if kind == ErrorKind::Ask {
-                            redirected = err.redirect_node().map(|x| format!("redis://{}", x.0));
+                            redirected = err
+                                .redirect_node()
+                                .map(|(node, _slot)| build_connection_string(node, None, self.tls));
                             is_asking = true;
                         } else if kind == ErrorKind::Moved {
                             // Refresh slots.
                             self.refresh_slots()?;
                             excludes.clear();
 
                             // Request again.
-                            redirected = err.redirect_node().map(|x| format!("redis://{}", x.0));
+                            redirected = err
+                                .redirect_node()
+                                .map(|(node, _slot)| build_connection_string(node, None, self.tls));
                             is_asking = false;
                             continue;
                         } else if kind == ErrorKind::TryAgain || kind == ErrorKind::ClusterDown {
@@ -448,7 +496,7 @@ impl ClusterConnection {
         let mut results = vec![Value::Nil; cmds.len()];
 
         let to_retry = self
-            .send_all_commands(&cmds)
+            .send_all_commands(cmds)
             .and_then(|node_cmds| self.recv_all_commands(&mut results, &node_cmds))?;
 
         if to_retry.is_empty() {
@@ -505,7 +553,7 @@ impl ClusterConnection {
         let mut cmd_map: HashMap<String, NodeCmd> = HashMap::new();
 
         for (idx, cmd) in cmds.iter().enumerate() {
-            let addr = self.get_addr_for_cmd(&cmd)?;
+            let addr = self.get_addr_for_cmd(cmd)?;
             let nc = cmd_map
                 .entry(addr.clone())
                 .or_insert_with(|| NodeCmd::new(addr));
@@ -682,7 +730,7 @@ fn get_random_connection<'a>(
 }
 
 // Get slot data from connection.
-fn get_slots(connection: &mut Connection) -> RedisResult<Vec<Slot>> {
+fn get_slots(connection: &mut Connection, tls_mode: Option<TlsMode>) -> RedisResult<Vec<Slot>> {
     let mut cmd = Cmd::new();
     cmd.arg("CLUSTER").arg("SLOTS");
     let value = connection.req_command(&cmd)?;
@@ -728,11 +776,11 @@ fn get_slots(connection: &mut Connection) -> RedisResult<Vec<Slot>> {
                         }
 
                         let port = if let Value::Int(port) = node[1] {
-                            port
+                            port as u16
                         } else {
                             return None;
                         };
-                        Some(format!("redis://{}:{}", ip, port))
+                        Some(build_connection_string(&ip, Some(port), tls_mode))
                     } else {
                         None
                     }
@@ -750,3 +798,17 @@ fn get_slots(connection: &mut Connection) -> RedisResult<Vec<Slot>> {
 
     Ok(result)
 }
+
+fn build_connection_string(host: &str, port: Option<u16>, tls_mode: Option<TlsMode>) -> String {
+    let host_port = match port {
+        Some(port) => format!("{}:{}", host, port),
+        None => host.to_string(),
+    };
+    match tls_mode {
+        None => format!("redis://{}", host_port),
+        Some(TlsMode::Insecure) => {
+            format!("rediss://{}/#insecure", host_port)
+        }
+        Some(TlsMode::Secure) => format!("rediss://{}", host_port),
+    }
+}

src/cluster_routing.rs

@@ -50,9 +50,9 @@ impl RoutingInfo {
     }
 
     pub fn for_key(key: &[u8]) -> Option<RoutingInfo> {
-        let key = match get_hashtag(&key) {
+        let key = match get_hashtag(key) {
             Some(tag) => tag,
-            None => &key,
+            None => key,
         };
         Some(RoutingInfo::Slot(
             crc16::State::<crc16::XMODEM>::calculate(key) % SLOT_SIZE as u16,

src/connection.rs

@@ -385,7 +385,12 @@ impl ActualConnection {
                 let tls = match timeout {
                     None => {
                         let tcp = TcpStream::connect((host, port))?;
-                        tls_connector.connect(host, tcp).unwrap()
+                        match tls_connector.connect(host, tcp) {
+                            Ok(res) => res,
+                            Err(e) => {
+                                fail!((ErrorKind::IoError, "SSL Handshake error", e.to_string()));
+                            }
+                        }
                     }
                     Some(timeout) => {
                         let mut tcp = None;
@@ -1114,7 +1119,7 @@ mod tests {
             ("tcp://127.0.0.1", false),
         ];
         for (url, expected) in cases.into_iter() {
-            let res = parse_redis_url(&url);
+            let res = parse_redis_url(url);
             assert_eq!(
                 res.is_some(),
                 expected,

tests/support/cluster.rs

@@ -2,18 +2,54 @@
 #![allow(dead_code)]
 
 use std::convert::identity;
-use std::fs;
+use std::env;
 use std::process;
 use std::thread::sleep;
 use std::time::Duration;
 
-use std::path::PathBuf;
+use tempfile::TempDir;
+
+use crate::support::build_keys_and_certs_for_tls;
 
 use super::RedisServer;
 
+const LOCALHOST: &str = "127.0.0.1";
+
+enum ClusterType {
+    Tcp,
+    TcpTls,
+}
+
+impl ClusterType {
+    fn get_intended() -> ClusterType {
+        match env::var("REDISRS_SERVER_TYPE")
+            .ok()
+            .as_ref()
+            .map(|x| &x[..])
+        {
+            Some("tcp") => ClusterType::Tcp,
+            Some("tcp+tls") => ClusterType::TcpTls,
+            val => {
+                panic!("Unknown server type {:?}", val);
+            }
+        }
+    }
+
+    fn build_addr(port: u16) -> redis::ConnectionAddr {
+        match ClusterType::get_intended() {
+            ClusterType::Tcp => redis::ConnectionAddr::Tcp("127.0.0.1".into(), port),
+            ClusterType::TcpTls => redis::ConnectionAddr::TcpTls {
+                host: "127.0.0.1".into(),
+                port,
+                insecure: true,
+            },
+        }
+    }
+}
+
 pub struct RedisCluster {
     pub servers: Vec<RedisServer>,
-    pub folders: Vec<PathBuf>,
+    pub folders: Vec<TempDir>,
 }
 
 impl RedisCluster {
@@ -22,25 +58,48 @@ impl RedisCluster {
         let mut folders = vec![];
         let mut addrs = vec![];
         let start_port = 7000;
+        let mut tls_paths = None;
+        let mut is_tls = false;
+
+        if let ClusterType::TcpTls = ClusterType::get_intended() {
+            // Create a shared set of keys in cluster mode
+            let tempdir = tempfile::Builder::new()
+                .prefix("redis")
+                .tempdir()
+                .expect("failed to create tempdir");
+            let files = build_keys_and_certs_for_tls(&tempdir);
+            folders.push(tempdir);
+            tls_paths = Some(files);
+            is_tls = true;
+        }
+
         for node in 0..nodes {
             let port = start_port + node;
 
             servers.push(RedisServer::new_with_addr(
-                redis::ConnectionAddr::Tcp("127.0.0.1".into(), port),
+                ClusterType::build_addr(port),
+                tls_paths.clone(),
                 |cmd| {
-                    let (a, b) = rand::random::<(u64, u64)>();
-                    let path = PathBuf::from(format!("/tmp/redis-rs-cluster-test-{}-{}-dir", a, b));
-                    fs::create_dir_all(&path).unwrap();
+                    let tempdir = tempfile::Builder::new()
+                        .prefix("redis")
+                        .tempdir()
+                        .expect("failed to create tempdir");
                     cmd.arg("--cluster-enabled")
                         .arg("yes")
                         .arg("--cluster-config-file")
-                        .arg(&path.join("nodes.conf"))
+                        .arg(&tempdir.path().join("nodes.conf"))
                         .arg("--cluster-node-timeout")
                         .arg("5000")
                         .arg("--appendonly")
                         .arg("yes");
-                    cmd.current_dir(&path);
-                    folders.push(path);
+                    if is_tls {
+                        cmd.arg("--tls-cluster").arg("yes");
+                        if replicas > 0 {
+                            cmd.arg("--tls-replication").arg("yes");
+                        }
+                    }
+                    cmd.current_dir(&tempdir.path());
+                    folders.push(tempdir);
                     addrs.push(format!("127.0.0.1:{}", port));
                     dbg!(&cmd);
                     cmd.spawn().unwrap()
@@ -59,6 +118,9 @@ impl RedisCluster {
             cmd.arg("--cluster-replicas").arg(replicas.to_string());
         }
         cmd.arg("--cluster-yes");
+        if is_tls {
+            cmd.arg("--tls").arg("--insecure");
+        }
         let status = dbg!(cmd).status().unwrap();
         assert!(status.success());
 
@@ -71,9 +133,15 @@ impl RedisCluster {
 
     fn wait_for_replicas(&self, replicas: u16) {
         'server: for server in &self.servers {
-            let addr = format!("redis://{}/", server.get_client_addr());
-            eprintln!("waiting until {} knows required number of replicas", addr);
-            let client = redis::Client::open(addr).unwrap();
+            let conn_info = redis::ConnectionInfo {
+                addr: server.get_client_addr().clone(),
+                redis: Default::default(),
+            };
+            eprintln!(
+                "waiting until {:?} knows required number of replicas",
+                conn_info.addr
+            );
+            let client = redis::Client::open(conn_info).unwrap();
             let mut con = client.get_connection().unwrap();
 
             // retry 100 times
@@ -98,9 +166,6 @@ impl RedisCluster {
         for server in &mut self.servers {
             server.stop();
         }
-        for folder in &self.folders {
-            fs::remove_dir_all(&folder).unwrap();
-        }
     }
 
     pub fn iter_servers(&self) -> impl Iterator<Item = &RedisServer> {
@@ -136,7 +201,10 @@ impl TestClusterContext {
         let mut builder = redis::cluster::ClusterClientBuilder::new(
             cluster
                 .iter_servers()
-                .map(|x| format!("redis://{}/", x.get_client_addr()))
+                .map(|server| redis::ConnectionInfo {
+                    addr: server.get_client_addr().clone(),
+                    redis: Default::default(),
+                })
                 .collect(),
         );
         builder = initializer(builder);