Repository: kafka-rust/kafka-rust (language: rust)

Issue:
Use latest rust-openssl
It would be nice to upgrade to rust-openssl 0.8.x (current stable branch) in order to avoid conflicts with other crates linking openssl 0.8.



Discussion / hints:
sounds good to me! have you got time to give this a try? i fear i might not be able to get my hands wet on this during this week.


Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index 9f1835288..f449012d3 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -20,7 +20,7 @@ fnv = "1.0"
 
 flate2 = { version = "0.2", optional = true }
 snappy = { version = "0.4", optional = true }
-openssl = { version = "0.7", optional = true }
+openssl = { version = "0.8", optional = true }
 
 
 [dev-dependencies]
diff --git a/examples/example-ssl.rs b/examples/example-ssl.rs
new file mode 100644
index 000000000..87ddc9967
--- /dev/null
+++ b/examples/example-ssl.rs
@@ -0,0 +1,153 @@
+fn main() {
+    example::main();
+}
+
+#[cfg(feature = "security")]
+mod example {
+    extern crate openssl;
+    extern crate kafka;
+    extern crate getopts;
+    extern crate env_logger;
+
+    use std::env;
+    use std::process;
+
+    use self::kafka::client::{FetchOffset, KafkaClient, SecurityConfig};
+
+    use self::openssl::ssl::{SslContext, SslMethod, SSL_VERIFY_PEER};
+    use self::openssl::x509::X509FileType;
+
+    pub fn main() {
+        env_logger::init().unwrap();
+
+        // ~ parse the command line arguments
+        let cfg = match Config::from_cmdline() {
+            Ok(cfg) => cfg,
+            Err(e) => {
+                println!("{}", e);
+                process::exit(1);
+            }
+        };
+
+        // ~ OpenSSL offers a variety of complex configurations. Here is an example:
+        let mut ctx = SslContext::new(SslMethod::Sslv23).unwrap();
+        ctx.set_cipher_list("DEFAULT").unwrap();
+        ctx.set_verify(SSL_VERIFY_PEER);
+        if let Some(ccert) = cfg.client_cert {
+            ctx.set_certificate_file(ccert, X509FileType::PEM).unwrap();
+        }
+        if let Some(ckey) = cfg.client_key {
+            ctx.set_private_key_file(ckey, X509FileType::PEM).unwrap();
+        }
+        if let Some(ca_cert) = cfg.ca_cert {
+            ctx.set_CA_file(ca_cert).unwrap();
+        } else {
+            // ~ allow client specify the CAs through the default paths:
+            // "These locations are read from the SSL_CERT_FILE and
+            // SSL_CERT_DIR environment variables if present, or defaults
+            // specified at OpenSSL build time otherwise."
+            ctx.set_default_verify_paths().unwrap();
+        }
+
+        // ~ instantiate KafkaClient with the previous OpenSSL setup
+        let mut client = KafkaClient::new_secure(cfg.brokers, SecurityConfig::new(ctx));
+
+        // ~ communicate with the brokers
+        match client.load_metadata_all() {
+            Err(e) => {
+                println!("{:?}", e);
+                drop(client);
+                process::exit(1);
+            }
+            Ok(_) => {
+                // ~ at this point we have successfully loaded
+                // metadata via a secured connection to one of the
+                // specified brokers
+
+                if client.topics().len() == 0 {
+                    println!("No topics available!");
+                } else {
+                    // ~ now let's communicate with all the brokers in
+                    // the cluster our topics are spread over
+
+                    let topics: Vec<String> = client.topics().names().map(Into::into).collect();
+                    match client.fetch_offsets(topics.as_slice(), FetchOffset::Latest) {
+                        Err(e) => {
+                            println!("{:?}", e);
+                            drop(client);
+                            process::exit(1);
+                        }
+                        Ok(toffsets) => {
+                            println!("Topic offsets:");
+                            for (topic, mut offs) in toffsets {
+                                offs.sort_by_key(|x| x.partition);
+                                println!("{}", topic);
+                                for off in offs {
+                                    println!("\t{}: {:?}", off.partition, off.offset);
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    struct Config {
+        brokers: Vec<String>,
+        client_cert: Option<String>,
+        client_key: Option<String>,
+        ca_cert: Option<String>,
+    }
+
+    impl Config {
+        fn from_cmdline() -> Result<Config, String> {
+            let mut opts = getopts::Options::new();
+            opts.optflag("h", "help", "Print this help screen");
+            opts.optopt("", "brokers", "Specify kafka brokers (comma separated)", "HOSTS");
+            opts.optopt("", "ca-cert", "Specify the trusted CA certificates", "FILE");
+            opts.optopt("", "client-cert", "Specify the client certificate", "FILE");
+            opts.optopt("", "client-key", "Specify key for the client certificate", "FILE");
+
+            let args: Vec<_> = env::args().collect();
+            let m = match opts.parse(&args[1..]) {
+                Ok(m) => m,
+                Err(e) => return Err(format!("{}", e)),
+            };
+
+            if m.opt_present("help") {
+                let brief = format!("{} [options]", args[0]);
+                return Err(opts.usage(&brief));
+            };
+
+            let brokers = m.opt_str("brokers")
+                .map(|s| {
+                    s.split(',')
+                        .map(|s| s.trim().to_owned())
+                        .filter(|s| !s.is_empty())
+                        .collect()
+                })
+                .unwrap_or_else(|| vec!["localhost:9092".into()]);
+            if brokers.is_empty() {
+                return Err("Invalid --brokers specified!".to_owned());
+            }
+
+            Ok(Config {
+                brokers: brokers,
+                client_cert: m.opt_str("client-cert"),
+                client_key: m.opt_str("client-key"),
+                ca_cert: m.opt_str("ca-cert"),
+            })
+        }
+    }
+}
+
+#[cfg(not(feature = "security"))]
+mod example {
+    use std::process;
+
+    pub fn main() {
+        println!("example relevant only with the \"security\" feature enabled!");
+        process::exit(1);
+    }
+}
diff --git a/src/client/mod.rs b/src/client/mod.rs
index 625149d20..be471bd70 100644
--- a/src/client/mod.rs
+++ b/src/client/mod.rs
@@ -408,7 +408,7 @@ impl KafkaClient {
     ///     ctx.set_certificate_file(&cert, X509FileType::PEM).unwrap();
     ///     ctx.set_private_key_file(&key, X509FileType::PEM).unwrap();
     ///     ctx.set_default_verify_paths().unwrap();
-    ///     ctx.set_verify(SSL_VERIFY_PEER, None);
+    ///     ctx.set_verify(SSL_VERIFY_PEER);
     ///
     ///     let mut client = KafkaClient::new_secure(vec!("localhost:9092".to_owned()),
     ///                                              SecurityConfig::new(ctx));
diff --git a/src/error.rs b/src/error.rs
index fb10fab6c..3d245dffe 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -6,7 +6,7 @@ use std::io;
 use std::fmt;
 
 #[cfg(feature = "security")]
-use openssl::ssl;
+use openssl::ssl::{self, Error as SslError};
 
 
 /// A type for results generated by this crate's functions where the `Err` type
@@ -25,7 +25,7 @@ pub enum Error {
     Kafka(KafkaCode),
     /// An error as reported by OpenSsl
     #[cfg(feature = "security")]
-    Ssl(ssl::error::SslError),
+    Ssl(SslError),
 
     /// Failure to correctly parse the server response due to the
     /// server speaking a newer protocol version (than the one this
@@ -180,32 +180,29 @@ impl From<io::Error> for Error {
 }
 
 #[cfg(feature = "security")]
-impl From<ssl::error::SslError> for Error {
-    fn from(err: ssl::error::SslError) -> Error {
+impl From<SslError> for Error {
+    fn from(err: SslError) -> Error {
         Error::Ssl(err)
     }
 }
 
+#[cfg(feature = "security")]
+impl<S> From<ssl::HandshakeError<S>> for Error {
+    fn from(err: ssl::HandshakeError<S>) -> Error {
+        match err {
+            ssl::HandshakeError::Failure(e) => From::from(e),
+            ssl::HandshakeError::Interrupted(s) => from_sslerror_ref(s.error()),
+        }
+    }
+}
+
 impl Clone for Error {
     fn clone(&self) -> Error {
         match self {
-            &Error::Io(ref err) => Error::Io(io::Error::new(err.kind(), "Io Error")),
+            &Error::Io(ref err) => Error::Io(clone_ioe(err)),
             &Error::Kafka(x) => Error::Kafka(x),
             #[cfg(feature = "security")]
-            &Error::Ssl(ref x) => {
-                match x {
-                    &ssl::error::SslError::StreamError(ref e) => {
-                        Error::Ssl(ssl::error::SslError::StreamError(io::Error::new(e.kind(),
-                                                                                    "Stream Error")))
-                    }
-                    &ssl::error::SslError::SslSessionClosed => {
-                        Error::Ssl(ssl::error::SslError::SslSessionClosed)
-                    }
-                    &ssl::error::SslError::OpenSslErrors(ref v) => {
-                        Error::Ssl(ssl::error::SslError::OpenSslErrors(v.clone()))
-                    }
-                }
-            }
+            &Error::Ssl(ref x) => from_sslerror_ref(x),
             &Error::UnsupportedProtocol => Error::UnsupportedProtocol,
             &Error::UnsupportedCompression => Error::UnsupportedCompression,
             #[cfg(feature = "snappy")]
@@ -220,6 +217,26 @@ impl Clone for Error {
     }
 }
 
+#[cfg(feature = "security")]
+fn from_sslerror_ref(err: &ssl::Error) -> Error {
+    match err {
+        &SslError::ZeroReturn => Error::Ssl(SslError::ZeroReturn),
+        &SslError::WantRead(ref e) => Error::Ssl(SslError::WantRead(clone_ioe(e))),
+        &SslError::WantWrite(ref e) => Error::Ssl(SslError::WantWrite(clone_ioe(e))),
+        &SslError::WantX509Lookup => Error::Ssl(SslError::WantX509Lookup),
+        &SslError::Stream(ref e) => Error::Ssl(SslError::Stream(clone_ioe(e))),
+        &SslError::Ssl(ref es) => Error::Ssl(SslError::Ssl(es.clone())),
+    }
+}
+
+/// Attempt to clone `io::Error`.
+fn clone_ioe(e: &io::Error) -> io::Error {
+    match e.raw_os_error() {
+        Some(code) => io::Error::from_raw_os_error(code),
+        None => io::Error::new(e.kind(), format!("Io error: {}", e)),
+    }
+}
+
 impl error::Error for Error {
     fn description(&self) -> &str {
         match *self {
@@ -243,6 +260,8 @@ impl error::Error for Error {
     fn cause(&self) -> Option<&error::Error> {
         match *self {
             Error::Io(ref err) => err.cause(),
+            #[cfg(feature = "security")]
+            Error::Ssl(ref err) => err.cause(),
             _ => None,
         }
     }


Test patch: