Add ssl.ca.certificate.stores

CHANGELOG.md

@@ -23,6 +23,9 @@ and the sticky consumer group partition assignor.
 
 ## Enhancements
 
+ * Windows: Added `ssl.ca.certificate.stores` to specify a list of
+ Windows Certificate Stores to read CA certificates from, e.g.,
+ `Intermediate,Root`. `Root` remains the default store.
  * Use reentrant `rand_r()` on supporting platforms which decreases lock
  contention (@azat).
  * Added `assignor` debug context for troubleshooting consumer partition

CONFIGURATION.md

@@ -69,6 +69,7 @@ ssl.certificate.pem | * | |
 ssl_certificate | * | | | low | Client's public key as set by rd_kafka_conf_set_ssl_cert() <br>*Type: see dedicated API*
 ssl.ca.location | * | | | low | File or directory path to CA certificate(s) for verifying the broker's key. Defaults: On Windows the system's CA certificates are automatically looked up in the Windows Root certificate store. On Mac OSX it is recommended to install openssl using Homebrew, to provide CA certificates. On Linux install the distribution's ca-certificates package. If OpenSSL is statically linked or `ssl.ca.location` is set to `probe` a list of standard paths will be probed and the first one found will be used as the default CA certificate location path. If OpenSSL is dynamically linked the OpenSSL library's default path will be used (see `OPENSSLDIR` in `openssl version -a`). <br>*Type: string*
 ssl_ca | * | | | low | CA certificate as set by rd_kafka_conf_set_ssl_cert() <br>*Type: see dedicated API*
+ssl.ca.certificate.stores | * | | Root | low | Comma-separated list of Windows Certificate stores to load CA certificates from. Certificates will be loaded in the same order as stores are specified. If no certificates can be loaded from any of the specified stores an error is logged and the OpenSSL library's default CA location is used instead. Store names are typically one or more of: MY, Root, Trust, CA. <br>*Type: string*
 ssl.crl.location | * | | | low | Path to CRL for verifying broker's certificate validity. <br>*Type: string*
 ssl.keystore.location | * | | | low | Path to client's keystore (PKCS#12) used for authentication. <br>*Type: string*
 ssl.keystore.password | * | | | low | Client's keystore (PKCS#12) password. <br>*Type: string*

INTRODUCTION.md

@@ -1047,7 +1047,9 @@ The CA root certificate defaults are system specific:
  CA path will be used, also called the OPENSSLDIR, which is typically
  `/etc/ssl/certs` (on Linux, typcially in the `ca-certificates` package) and
  `/usr/local/etc/openssl` on Mac OSX (Homebrew).
- * On Windows the Root certificate store is used.
+ * On Windows the Root certificate store is used, unless
+ `ssl.ca.certificate.stores` is configured in which case certificates are
+ read from the specified stores.
  * If OpenSSL is linked statically, librdkafka will set the default CA
  location to the first of a series of probed paths (see below).
 
@@ -1085,6 +1087,21 @@ used when OpenSSL is statically linked:
  etc..
 
 
+On **Windows** the Root certificate store is read by default, but any number
+of certificate stores can be read by setting the `ssl.ca.certificate.stores`
+configuration property to a comma-separated list of certificate store names.
+The predefined system store names are:
+
+ * `MY` - User certificates
+ * `Root` - System CA certificates (default)
+ * `CA` - Intermediate CA certificates
+ * `Trust` - Trusted publishers
+
+For example, to read both intermediate and root CAs, set
+`ssl.ca.certificate.stores=CA,Root`.
+
+
+
 #### Sparse connections
 
 The client will only connect to brokers it needs to communicate with, and

src/rdkafka_conf.c

@@ -789,6 +789,20 @@ static const struct rd_kafka_property rd_kafka_properties[] = {
  .copy = rd_kafka_conf_cert_copy,
  _UNSUPPORTED_SSL
  },
+ { _RK_GLOBAL, "ssl.ca.certificate.stores", _RK_C_STR,
+ _RK(ssl.ca_cert_stores),
+ "Comma-separated list of Windows Certificate stores to load "
+ "CA certificates from. Certificates will be loaded in the same "
+ "order as stores are specified. If no certificates can be loaded "
+ "from any of the specified stores an error is logged and the "
+ "OpenSSL library's default CA location is used instead. "
+ "Store names are typically one or more of: MY, Root, Trust, CA.",
+ .sdef = "Root",
+#if !defined(_WIN32)
+ .unsupported = "configuration only valid on Windows"
+#endif
+ },
+
  { _RK_GLOBAL, "ssl.crl.location", _RK_C_STR,
  _RK(ssl.crl_location),
  "Path to CRL for verifying broker's certificate validity.",

src/rdkafka_conf.h

@@ -235,6 +235,8 @@ struct rd_kafka_conf_s {
  rd_kafka_cert_t *cert;
  char *ca_location;
  rd_kafka_cert_t *ca;
+ /** CSV list of Windows certificate stores */
+ char *ca_cert_stores;
  char *crl_location;
  char *keystore_location;
  char *keystore_password;

src/rdkafka_ssl.c

@@ -42,6 +42,8 @@
 
 #include <openssl/x509.h>
 
+#include <ctype.h>
+
 #if !_WIN32
 #include <sys/types.h>
 #include <sys/stat.h>
@@ -692,25 +694,45 @@ static X509 *rd_kafka_ssl_X509_from_string (rd_kafka_t *rk, const char *str) {
 #ifdef _WIN32
 
 /**
- * @brief Attempt load CA certificates from the Windows Certificate Root store.
+ * @brief Attempt load CA certificates from a Windows Certificate store.
  */
-static int rd_kafka_ssl_win_load_root_certs (rd_kafka_t *rk, SSL_CTX *ctx) {
+static int rd_kafka_ssl_win_load_cert_store (rd_kafka_t *rk, SSL_CTX *ctx,
+ const char *store_name) {
  HCERTSTORE w_store;
  PCCERT_CONTEXT w_cctx = NULL;
  X509_STORE *store;
  int fail_cnt = 0, cnt = 0;
  char errstr[256];
+ wchar_t *wstore_name;
+ size_t wsize = 0;
+ errno_t werr;
+
+ /* Convert store_name to wide-char */
+ werr = mbstowcs_s(&wsize, NULL, 0, store_name, strlen(store_name));
+ if (werr || wsize < 2 || wsize > 1000) {
+ rd_kafka_log(rk, LOG_ERR, "CERTSTORE",
+ "Invalid Windows certificate store name: %.*s%s",
+ 30, store_name,
+ wsize < 2 ? " (empty)" : " (truncated)");
+ return -1;
+ }
+ wstore_name = rd_alloca(sizeof(*wstore_name) * wsize);
+ werr = mbstowcs_s(NULL, wstore_name, wsize, store_name,
+ strlen(store_name));
+ rd_assert(!werr);
 
  w_store = CertOpenStore(CERT_STORE_PROV_SYSTEM,
  0,
  0,
- CERT_SYSTEM_STORE_CURRENT_USER,
- L"Root");
+ CERT_SYSTEM_STORE_CURRENT_USER|
+ CERT_STORE_READONLY_FLAG|
+ CERT_STORE_OPEN_EXISTING_FLAG,
+ wstore_name);
  if (!w_store) {
- rd_kafka_dbg(rk, SECURITY, "CERTROOT",
+ rd_kafka_log(rk, LOG_ERR, "CERTSTORE",
  "Failed to open Windows certificate "
- "Root store: %s: "
- "falling back to OpenSSL default CA paths",
+ "%s store: %s",
+ store_name,
  rd_strerror_w32(GetLastError(), errstr,
  sizeof(errstr)));
  return -1;
@@ -723,8 +745,6 @@ static int rd_kafka_ssl_win_load_root_certs (rd_kafka_t *rk, SSL_CTX *ctx) {
  while ((w_cctx = CertEnumCertificatesInStore(w_store, w_cctx))) {
  X509 *x509;
 
- cnt++;
-
  /* Parse Windows cert: DER -> X.509 */
  x509 = d2i_X509(NULL,
  (const unsigned char **)&w_cctx->pbCertEncoded,
@@ -736,7 +756,9 @@ static int rd_kafka_ssl_win_load_root_certs (rd_kafka_t *rk, SSL_CTX *ctx) {
 
  /* Add cert to OpenSSL's trust store */
  if (!X509_STORE_add_cert(store, x509))
- fail_cnt++;
+ fail_cnt++;
+ else
+ cnt++;
 
  X509_free(x509);
  }
@@ -746,16 +768,74 @@ static int rd_kafka_ssl_win_load_root_certs (rd_kafka_t *rk, SSL_CTX *ctx) {
 
  CertCloseStore(w_store, 0);
 
- rd_kafka_dbg(rk, SECURITY, "CERTROOT",
- "%d/%d certificate(s) successfully added from "
- "Windows Certificate Root store",
- cnt - fail_cnt, cnt);
+ rd_kafka_dbg(rk, SECURITY, "CERTSTORE",
+ "%d certificate(s) successfully added from "
+ "Windows Certificate %s store, %d failed",
+ cnt, store_name, fail_cnt);
 
- return cnt - fail_cnt == 0 ? -1 : 0;
+ if (cnt == 0 && fail_cnt > 0)
+ return -1;
+
+ return cnt;
+}
+
+/**
+ * @brief Load certs from the configured CSV list of Windows Cert stores.
+ *
+ * @returns the number of successfully loaded certificates, or -1 on error.
+ */
+static int rd_kafka_ssl_win_load_cert_stores (rd_kafka_t *rk,
+ SSL_CTX *ctx,
+ const char *store_names) {
+ char *s;
+ int cert_cnt = 0, fail_cnt = 0;
+
+ if (!store_names || !*store_names)
+ return 0;
+
+ rd_strdupa(&s, store_names);
+
+ /* Parse CSV list ("Root,CA, , ,Something") and load
+ * each store in order. */
+ while (*s) {
+ char *t;
+ const char *store_name;
+ int r;
+
+ while (isspace((int)*s) || *s == ',')
+ s++;
+
+ if (!*s)
+ break;
+
+ store_name = s;
+
+ t = strchr(s, (int)',');
+ if (t) {
+ *t = '\0';
+ s = t+1;
+ for (; t >= store_name && isspace((int)*t) ; t--)
+ *t = '\0';
+ } else {
+ s = "";
+ }
+
+ r = rd_kafka_ssl_win_load_cert_store(rk, ctx, store_name);
+ if (r != -1)
+ cert_cnt += r;
+ else
+ fail_cnt++;
+ }
+
+ if (cert_cnt == 0 && fail_cnt > 0)
+ return -1;
+
+ return cert_cnt;
 }
 #endif /* MSC_VER */
 
 
+
 /**
  * @brief Probe for the system's CA certificate location and if found set it
  * on the \p CTX.
@@ -908,8 +988,20 @@ static int rd_kafka_ssl_set_certs (rd_kafka_t *rk, SSL_CTX *ctx,
  } else {
 #ifdef _WIN32
  /* Attempt to load CA root certificates from the
- * Windows crypto Root cert store. */
- r = rd_kafka_ssl_win_load_root_certs(rk, ctx);
+ * configured Windows certificate stores. */
+ r = rd_kafka_ssl_win_load_cert_stores(
+ rk, ctx, rk->rk_conf.ssl.ca_cert_stores);
+ if (r == 0) {
+ rd_kafka_log(rk, LOG_NOTICE, "CERTSTORE",
+ "No CA certificates loaded from "
+ "Windows certificate stores: "
+ "falling back to default OpenSSL CA paths");
+ r = -1;
+ } else if (r == -1)
+ rd_kafka_log(rk, LOG_NOTICE, "CERTSTORE",
+ "Failed to load CA certificates from "
+ "Windows certificate stores: "
+ "falling back to default OpenSSL CA paths");
 #else
  r = -1;
 #endif

tests/0004-conf.c

@@ -380,6 +380,9 @@ int main_0004_conf (int argc, char **argv) {
  "auto.offset.reset", "earliest", /* Global->Topic fallthru */
 #if WITH_ZLIB
  "compression.codec", "gzip", /* S2I property */
+#endif
+#if defined(_WIN32)
+ "ssl.ca.certificate.stores", "Intermediate ,, Root ,",
 #endif
  NULL
  };