Read all certificates from ssl.ca.pem, not just the first one (#4049)

* Read all certificates from ssl.ca.pem, not just the first one

Reported in confluentinc/confluent-kafka-go#827

* Fix memory leak when reading DER-encoded certificates

Author: edenhill

CHANGELOG.md

@@ -50,6 +50,7 @@ configuration property.
 
  * Windows: couldn't read a PKCS#12 keystore correctly because binary mode wasn't explicitly set and Windows defaults to text mode.
  * Fixed memory leak when loading SSL certificates (@Mekk, #3930)
+ * Load all CA certificates from `ssl.ca.pem`, not just the first one.
 
 ### Transactional producer fixes
 

src/rdkafka_cert.c

@@ -253,6 +253,8 @@ static rd_kafka_cert_t *rd_kafka_cert_new(const rd_kafka_conf_t *conf,
                                 X509_free(x509);
                                 goto fail;
                         }
+
+                        X509_free(x509);
                 } break;
 
                 case RD_KAFKA_CERT_ENC_PEM: {

src/rdkafka_ssl.c

@@ -1019,32 +1019,55 @@ static int rd_kafka_ssl_set_certs(rd_kafka_t *rk,
                         /* CA as PEM string */
                         X509 *x509;
                         X509_STORE *store;
+                        BIO *bio;
+                        int cnt = 0;
 
                         /* Get the OpenSSL trust store */
                         store = SSL_CTX_get_cert_store(ctx);
                         rd_assert(store != NULL);
 
                         rd_kafka_dbg(rk, SECURITY, "SSL",
-                                     "Loading CA certificate from string");
+                                     "Loading CA certificate(s) from string");
+
+                        bio =
+                            BIO_new_mem_buf((void *)rk->rk_conf.ssl.ca_pem, -1);
+                        rd_assert(bio != NULL);
+
+                        /* Add all certificates to cert store */
+                        while ((x509 = PEM_read_bio_X509(
+                                    bio, NULL, rd_kafka_transport_ssl_passwd_cb,
+                                    rk))) {
+                                if (!X509_STORE_add_cert(store, x509)) {
+                                        rd_snprintf(errstr, errstr_size,
+                                                    "failed to add ssl.ca.pem "
+                                                    "certificate "
+                                                    "#%d to CA cert store: ",
+                                                    cnt);
+                                        X509_free(x509);
+                                        BIO_free(bio);
+                                        return -1;
+                                }
 
-                        x509 = rd_kafka_ssl_X509_from_string(
-                            rk, rk->rk_conf.ssl.ca_pem);
-                        if (!x509) {
-                                rd_snprintf(errstr, errstr_size,
-                                            "ssl.ca.pem failed: "
-                                            "not in PEM format?: ");
-                                return -1;
+                                X509_free(x509);
+                                cnt++;
                         }
 
-                        if (!X509_STORE_add_cert(store, x509)) {
+                        if (!BIO_eof(bio) || !cnt) {
                                 rd_snprintf(errstr, errstr_size,
-                                            "failed to add ssl.ca.pem to "
-                                            "CA cert store: ");
-                                X509_free(x509);
+                                            "failed to read certificate #%d "
+                                            "from ssl.ca.pem: "
+                                            "not in PEM format?: ",
+                                            cnt);
+                                BIO_free(bio);
                                 return -1;
                         }
 
-                        X509_free(x509);
+                        BIO_free(bio);
+
+                        rd_kafka_dbg(rk, SECURITY, "SSL",
+                                     "Loaded %d CA certificate(s) from string",
+                                     cnt);
+
 
                         ca_probe = rd_false;
                 }

tests/0097-ssl_verify.cpp

@@ -51,7 +51,7 @@ static const std::string envname[RdKafka::CERT__CNT][RdKafka::CERT_ENC__CNT] = {
     {
         "SSL_pkcs",
         "SSL_ca_der",
-        "SSL_ca_pem",
+        "SSL_all_cas_pem" /* Contains multiple CA certs */,
     }};
 
 
@@ -118,26 +118,45 @@ class TestVerifyCb : public RdKafka::SslCertificateVerifyCb {
 };
 
 
+/**
+ * @brief Set SSL PEM cert/key using configuration property.
+ *
+ * The cert/key is loadded from environment variables set up by trivup.
+ *
+ * @param loc_prop ssl.X.location property that will be cleared.
+ * @param pem_prop ssl.X.pem property that will be set.
+ * @param cert_type Certificate type.
+ */
 static void conf_location_to_pem(RdKafka::Conf *conf,
                                  std::string loc_prop,
-                                 std::string pem_prop) {
+                                 std::string pem_prop,
+                                 RdKafka::CertificateType cert_type) {
   std::string loc;
 
-
-  if (conf->get(loc_prop, loc) != RdKafka::Conf::CONF_OK)
-    Test::Fail("Failed to get " + loc_prop);
-
   std::string errstr;
   if (conf->set(loc_prop, "", errstr) != RdKafka::Conf::CONF_OK)
     Test::Fail("Failed to reset " + loc_prop + ": " + errstr);
 
+  const char *p;
+  p = test_getenv(envname[cert_type][RdKafka::CERT_ENC_PEM].c_str(), NULL);
+  if (!p)
+    Test::Fail(
+        "Invalid test environment: "
+        "Missing " +
+        envname[cert_type][RdKafka::CERT_ENC_PEM] +
+        " env variable: make sure trivup is up to date");
+
+  loc = p;
+
+
   /* Read file */
   std::ifstream ifs(loc.c_str());
   std::string pem((std::istreambuf_iterator<char>(ifs)),
                   std::istreambuf_iterator<char>());
 
-  Test::Say("Read " + loc_prop + "=" + loc +
-            " from disk and changed to in-memory " + pem_prop + "\n");
+  Test::Say("Read env " + envname[cert_type][RdKafka::CERT_ENC_PEM] + "=" +
+            loc + " from disk and changed to in-memory " + pem_prop +
+            " string\n");
 
   if (conf->set(pem_prop, pem, errstr) != RdKafka::Conf::CONF_OK)
     Test::Fail("Failed to set " + pem_prop + ": " + errstr);
@@ -178,7 +197,8 @@ static void conf_location_to_setter(RdKafka::Conf *conf,
   loc = p;
 
   Test::Say(tostr() << "Reading " << loc_prop << " file " << loc << " as "
-                    << encnames[encoding] << "\n");
+                    << encnames[encoding] << " from env "
+                    << envname[cert_type][encoding] << "\n");
 
   /* Read file */
   std::ifstream ifs(loc.c_str(), std::ios::binary | std::ios::ate);
@@ -200,8 +220,8 @@ static void conf_location_to_setter(RdKafka::Conf *conf,
 
 
 typedef enum {
-  USE_LOCATION, /* use ssl.key.location */
-  USE_CONF,     /* use ssl.key.pem */
+  USE_LOCATION, /* use ssl.X.location */
+  USE_CONF,     /* use ssl.X.pem */
   USE_SETTER,   /* use conf->set_ssl_cert(), this supports multiple formats */
 } cert_load_t;
 
@@ -245,20 +265,22 @@ static void do_test_verify(const int line,
   /* Get ssl.key.location, read its contents, and replace with
    * ssl.key.pem. Same with ssl.certificate.location -> ssl.certificate.pem. */
   if (load_key == USE_CONF)
-    conf_location_to_pem(conf, "ssl.key.location", "ssl.key.pem");
+    conf_location_to_pem(conf, "ssl.key.location", "ssl.key.pem",
+                         RdKafka::CERT_PRIVATE_KEY);
   else if (load_key == USE_SETTER)
     conf_location_to_setter(conf, "ssl.key.location", RdKafka::CERT_PRIVATE_KEY,
                             key_enc);
 
   if (load_pub == USE_CONF)
     conf_location_to_pem(conf, "ssl.certificate.location",
-                         "ssl.certificate.pem");
+                         "ssl.certificate.pem", RdKafka::CERT_PUBLIC_KEY);
   else if (load_pub == USE_SETTER)
     conf_location_to_setter(conf, "ssl.certificate.location",
                             RdKafka::CERT_PUBLIC_KEY, pub_enc);
 
   if (load_ca == USE_CONF)
-    conf_location_to_pem(conf, "ssl.ca.location", "ssl.ca.pem");
+    conf_location_to_pem(conf, "ssl.ca.location", "ssl.ca.pem",
+                         RdKafka::CERT_CA);
   else if (load_ca == USE_SETTER)
     conf_location_to_setter(conf, "ssl.ca.location", RdKafka::CERT_CA, ca_enc);
 
@@ -391,6 +413,12 @@ int main_0097_ssl_verify(int argc, char **argv) {
   do_test_verify(__LINE__, true, USE_LOCATION, RdKafka::CERT_ENC_PEM,
                  USE_SETTER, RdKafka::CERT_ENC_DER, USE_SETTER,
                  RdKafka::CERT_ENC_DER);
+  do_test_verify(__LINE__, true, USE_LOCATION, RdKafka::CERT_ENC_PEM,
+                 USE_SETTER, RdKafka::CERT_ENC_DER, USE_SETTER,
+                 RdKafka::CERT_ENC_PEM); /* env: SSL_all_cas_pem */
+  do_test_verify(__LINE__, true, USE_LOCATION, RdKafka::CERT_ENC_PEM,
+                 USE_SETTER, RdKafka::CERT_ENC_DER, USE_CONF,
+                 RdKafka::CERT_ENC_PEM); /* env: SSL_all_cas_pem */
   do_test_verify(__LINE__, true, USE_SETTER, RdKafka::CERT_ENC_PKCS12,
                  USE_SETTER, RdKafka::CERT_ENC_PKCS12, USE_SETTER,
                  RdKafka::CERT_ENC_PKCS12);

tests/requirements.txt

@@ -1,2 +1,2 @@
-trivup >= 0.11.0
+trivup >= 0.12.0
 jsoncomment