Improve HTTPS CA certificates configuration (#5107)

Use `rd_kafka_ssl_probe_and_set_default_ca_location`  and set same CA search policy as Kafka SSL.

- on Windows try to load from certificate store or fallback to paths in case using msys2.
- on macOS use `probe` by default.
- on Linux use `probe` when configured, check default path for dynamic linked OpenSSL or fallback to `probe` when statically linked.


---------

Co-authored-by: Emanuele Sabellico <[email protected]>

Author: anchitj

CHANGELOG.md

@@ -9,6 +9,9 @@ librdkafka v2.11.0 is a feature release:
   initialized (#4718).
 * Features BROKER_BALANCED_CONSUMER and SASL_GSSAPI don't depend on
   JoinGroup v0 anymore, missing in AK 4.0 and CP 8.0 (#5131).
+* Improve HTTPS CA certificates configuration by probing several paths
+  when OpenSSL is statically linked and providing a way to customize their location
+  or value (#).
 
 
 ## Fixes

CONFIGURATION.md

@@ -73,6 +73,8 @@ ssl.certificate.location                 |  *  |                 |
 ssl.certificate.pem                      |  *  |                 |               | low        | Client's public key string (PEM format) used for authentication. <br>*Type: string*
 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 this configuration defaults to `probe`. 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*
+https.ca.location                        |  *  |                 |               | low        | File or directory path to CA certificate(s) for verifying HTTPS endpoints, like `sasl.oauthbearer.token.endpoint.url` used for OAUTHBEARER/OIDC authentication. Mutually exclusive with `https.ca.pem`. Defaults: On Windows the system's CA certificates are automatically looked up in the Windows Root certificate store. On Mac OSX this configuration defaults to `probe`. 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 `https.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*
+https.ca.pem                             |  *  |                 |               | low        | CA certificate string (PEM format) for verifying HTTPS endpoints. Mutually exclusive with `https.ca.location`. Optional: see `https.ca.location`. <br>*Type: string*
 ssl.ca.pem                               |  *  |                 |               | low        | CA certificate string (PEM format) for verifying the broker's key. <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*

INTRODUCTION.md

@@ -1197,7 +1197,40 @@ To use this authentication method the client needs to be configured as follows:
   provided to the broker. A comma-separated list of key=value pairs.
   For example:
   `supportFeatureX=true,organizationId=sales-emea`
-
+* `https.ca.location` - (optional) to customize the CA certificates
+   location.
+
+* `https.ca.pem` - (optional) to provide the CA certificates as a PEM string.
+
+##### JWT bearer grant type (KIP-1139)
+
+This KIP adds support for the `client_credentials, urn:ietf:params:oauth:grant-type:jwt-bearer`
+grant type, with a series of properties to be used for creating a JWT assertion
+sent to the token endpoint. The authenticated principal corresponds to the
+`sub` claim returned by token endpoint, `sasl.oauthbearer.client.id` and
+`sasl.oauthbearer.client.secret` aren't used. Required JWT claims must be set
+either through the template or with the `claim` properties.
+
+* `sasl.oauthbearer.grant.type` - changes the default grant type, set it to
+  `urn:ietf:params:oauth:grant-type:jwt-bearer`.
+* `sasl.oauthbearer.assertion.algorithm` - JWT algorithm defaults to `RS256`.
+* `sasl.oauthbearer.assertion.private.key.file` - a private key file for signing
+   the token.
+* `sasl.oauthbearer.assertion.private.key.passphrase` - (optional) passphrase for the key if encrypted.
+* `sasl.oauthbearer.assertion.private.key.pem` - alternatively to the key file
+  it's possible to pass the private key as a string.
+* `sasl.oauthbearer.assertion.file` - (optional) assertion file: with this property all other
+  assertion related fields are ignored and the assertion is read from this file
+  that should be periodically updated.
+* `sasl.oauthbearer.assertion.jwt.template.file` - (optional) template file: a template containing
+  a default `header` and `payload` that can be overwritten by the `claim` properties.
+* `sasl.oauthbearer.assertion.claim.aud`,
+  `sasl.oauthbearer.assertion.claim.exp.seconds`,
+  `sasl.oauthbearer.assertion.claim.iss`,
+  `sasl.oauthbearer.assertion.claim.jti.include`,
+  `sasl.oauthbearer.assertion.claim.sub` - (optional) the `claim` properties:
+  it's possible to dynamically customize the JWT claims with these or to
+  skip the template file and use only these properties.
 
 <a name="sparse-connections"></a>
 #### Sparse connections

src/rd.h

@@ -53,6 +53,7 @@
 #include <time.h>
 #include <assert.h>
 #include <limits.h>
+#include <sys/stat.h>
 
 #include "tinycthread.h"
 #include "rdsysqueue.h"
@@ -544,4 +545,34 @@ rd_file_mkstemp(const char *prefix,
         return tempfile;
 }
 
+/**
+ * @brief Retrive stat for a \p path .
+ *
+ * @param path Path to the file or directory.
+ * @param is_dir Pointer to store if the \p path is a directory (optional).
+ *
+ * @return `rd_true` if the path exists.
+ */
+static RD_INLINE RD_UNUSED rd_bool_t rd_file_stat(const char *path,
+                                                  rd_bool_t *is_dir) {
+#ifdef _WIN32
+        struct _stat st;
+        if (_stat(path, &st) == 0) {
+                if (is_dir)
+                        *is_dir = st.st_mode & S_IFDIR;
+                return rd_true;
+        }
+#else
+        struct stat st;
+        if (stat(path, &st) == 0) {
+                if (is_dir)
+                        *is_dir = S_ISDIR(st.st_mode);
+                return rd_true;
+        }
+#endif
+        if (is_dir)
+                *is_dir = rd_false;
+        return rd_false;
+}
+
 #endif /* _RD_H_ */

src/rdhttp.c

@@ -40,6 +40,10 @@
 #include <curl/curl.h>
 #include "rdhttp.h"
 
+#if WITH_SSL
+#include "rdkafka_ssl.h"
+#endif
+
 /** Maximum response size, increase as necessary. */
 #define RD_HTTP_RESPONSE_SIZE_MAX 1024 * 1024 * 500 /* 500kb */
 
@@ -128,8 +132,143 @@ rd_http_req_write_cb(char *ptr, size_t size, size_t nmemb, void *userdata) {
         return nmemb;
 }
 
-rd_http_error_t *rd_http_req_init(rd_http_req_t *hreq, const char *url) {
+#if WITH_SSL
+/**
+ * @brief Callback function for setting up the SSL_CTX for HTTPS requests.
+ *
+ * This function sets the default CA paths for the SSL_CTX, and if that fails,
+ * it attempts to probe and set a default CA location. If `probe` is forced
+ * it skips the default CA paths and directly probes for CA certificates.
+ *
+ * On Windows, it attempts to load CA root certificates from the
+ * configured Windows certificate stores before falling back to the default.
+ *
+ * @return `CURLE_OK` on success, or `CURLE_SSL_CACERT_BADFILE` on failure.
+ */
+static CURLcode
+rd_http_ssl_ctx_function(CURL *curl, void *sslctx, void *userptr) {
+        SSL_CTX *ctx   = (SSL_CTX *)sslctx;
+        rd_kafka_t *rk = (rd_kafka_t *)userptr;
+        int r          = -1;
+        rd_bool_t force_probe =
+            !rd_strcmp(rk->rk_conf.https.ca_location, "probe");
+        rd_bool_t use_probe = force_probe;
+
+#if WITH_STATIC_LIB_libcrypto
+        /* We fallback to `probe` when statically linked. */
+        use_probe = rd_true;
+#endif
+
+#ifdef _WIN32
+        /* Attempt to load CA root certificates from the
+         * configured Windows certificate stores. */
+        r = rd_kafka_ssl_win_load_cert_stores(rk, "https", ctx,
+                                              rk->rk_conf.ssl.ca_cert_stores);
+        if (r == 0) {
+                rd_kafka_log(rk, LOG_NOTICE, "CERTSTORE",
+                             "No CA certificates loaded for `https` 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 for `https` from "
+                             "Windows certificate stores: "
+                             "falling back to default OpenSSL CA paths");
+
+        if (r != -1) {
+                rd_kafka_dbg(rk, SECURITY, "SSL",
+                             "Successfully loaded CA certificates for `https` "
+                             "from Windows certificate stores");
+                return CURLE_OK; /* Success, CA certs loaded on Windows */
+        }
+#endif
 
+        if (!force_probe) {
+                /* Previous default behavior: use predefined paths set when
+                 * building OpenSSL. */
+                char errstr[512];
+                r = SSL_CTX_set_default_verify_paths(ctx);
+                if (r == 1) {
+                        rd_kafka_dbg(rk, SECURITY, "SSL",
+                                     "SSL_CTX_set_default_verify_paths() "
+                                     "for `https` "
+                                     "succeeded");
+                        return CURLE_OK; /* Success */
+                }
+
+                /* Read error and clear the error stack. */
+                rd_kafka_ssl_error0(rk, NULL, "https", errstr, sizeof(errstr));
+                rd_kafka_dbg(rk, SECURITY, "SSL",
+                             "SSL_CTX_set_default_verify_paths() "
+                             "for `https` "
+                             "failed: %s",
+                             errstr);
+        }
+
+        if (use_probe) {
+                /* We asked for probing or we're using
+                 * a statically linked version of OpenSSL. */
+
+                r = rd_kafka_ssl_probe_and_set_default_ca_location(rk, "https",
+                                                                   ctx);
+                if (r == 0)
+                        return CURLE_OK;
+        }
+
+        return CURLE_SSL_CACERT_BADFILE;
+}
+
+static void rd_http_ssl_configure(rd_kafka_t *rk, CURL *hreq_curl) {
+        rd_bool_t force_probe =
+            !rd_strcmp(rk->rk_conf.https.ca_location, "probe");
+
+        if (!force_probe && rk->rk_conf.https.ca_location) {
+                rd_bool_t is_dir;
+                rd_kafka_dbg(rk, SECURITY, "SSL",
+                             "Setting `https` CA certs from "
+                             "configured location: %s",
+                             rk->rk_conf.https.ca_location);
+                if (rd_file_stat(rk->rk_conf.https.ca_location, &is_dir)) {
+                        if (is_dir) {
+                                curl_easy_setopt(hreq_curl, CURLOPT_CAPATH,
+                                                 rk->rk_conf.https.ca_location);
+                                curl_easy_setopt(hreq_curl, CURLOPT_CAINFO,
+                                                 NULL);
+                        } else {
+                                curl_easy_setopt(hreq_curl, CURLOPT_CAPATH,
+                                                 NULL);
+                                curl_easy_setopt(hreq_curl, CURLOPT_CAINFO,
+                                                 rk->rk_conf.https.ca_location);
+                        }
+                } else {
+                        /* Path doesn't exist, don't set any trusted
+                         * certificate. */
+                        curl_easy_setopt(hreq_curl, CURLOPT_CAINFO, NULL);
+                        curl_easy_setopt(hreq_curl, CURLOPT_CAPATH, NULL);
+                }
+        } else if (!force_probe && rk->rk_conf.https.ca_pem) {
+                struct curl_blob ca_blob = {
+                    .data  = rk->rk_conf.https.ca_pem,
+                    .len   = strlen(rk->rk_conf.https.ca_pem),
+                    .flags = CURL_BLOB_COPY};
+                rd_kafka_dbg(rk, SECURITY, "SSL",
+                             "Setting `https` CA certs from "
+                             "configured PEM string");
+                curl_easy_setopt(hreq_curl, CURLOPT_CAINFO_BLOB, &ca_blob);
+                /* Only the blob should be set, no default paths. */
+                curl_easy_setopt(hreq_curl, CURLOPT_CAINFO, NULL);
+                curl_easy_setopt(hreq_curl, CURLOPT_CAPATH, NULL);
+        } else {
+                curl_easy_setopt(hreq_curl, CURLOPT_SSL_CTX_FUNCTION,
+                                 rd_http_ssl_ctx_function);
+                curl_easy_setopt(hreq_curl, CURLOPT_SSL_CTX_DATA, rk);
+        }
+}
+#endif
+
+rd_http_error_t *
+rd_http_req_init(rd_kafka_t *rk, rd_http_req_t *hreq, const char *url) {
         memset(hreq, 0, sizeof(*hreq));
 
         hreq->hreq_curl = curl_easy_init();
@@ -157,6 +296,10 @@ rd_http_error_t *rd_http_req_init(rd_http_req_t *hreq, const char *url) {
                          rd_http_req_write_cb);
         curl_easy_setopt(hreq->hreq_curl, CURLOPT_WRITEDATA, (void *)hreq);
 
+#if WITH_SSL
+        rd_http_ssl_configure(rk, hreq->hreq_curl);
+#endif
+
         return NULL;
 }
 
@@ -207,13 +350,14 @@ const char *rd_http_req_get_content_type(rd_http_req_t *hreq) {
  * by calling rd_http_error_destroy(). In case of HTTP error the \p *rbufp
  * may be filled with the error response.
  */
-rd_http_error_t *rd_http_get(const char *url, rd_buf_t **rbufp) {
+rd_http_error_t *
+rd_http_get(rd_kafka_t *rk, const char *url, rd_buf_t **rbufp) {
         rd_http_req_t hreq;
         rd_http_error_t *herr;
 
         *rbufp = NULL;
 
-        herr = rd_http_req_init(&hreq, url);
+        herr = rd_http_req_init(rk, &hreq, url);
         if (unlikely(herr != NULL))
                 return herr;
 
@@ -317,7 +461,7 @@ rd_http_error_t *rd_http_post_expect_json(rd_kafka_t *rk,
         size_t len;
         const char *content_type;
 
-        herr = rd_http_req_init(&hreq, url);
+        herr = rd_http_req_init(rk, &hreq, url);
         if (unlikely(herr != NULL))
                 return herr;
 
@@ -382,7 +526,8 @@ rd_http_error_t *rd_http_post_expect_json(rd_kafka_t *rk,
  *
  * Same error semantics as rd_http_get().
  */
-rd_http_error_t *rd_http_get_json(const char *url, cJSON **jsonp) {
+rd_http_error_t *
+rd_http_get_json(rd_kafka_t *rk, const char *url, cJSON **jsonp) {
         rd_http_req_t hreq;
         rd_http_error_t *herr;
         rd_slice_t slice;
@@ -393,7 +538,7 @@ rd_http_error_t *rd_http_get_json(const char *url, cJSON **jsonp) {
 
         *jsonp = NULL;
 
-        herr = rd_http_req_init(&hreq, url);
+        herr = rd_http_req_init(rk, &hreq, url);
         if (unlikely(herr != NULL))
                 return herr;
 
@@ -468,19 +613,21 @@ int unittest_http(void) {
         cJSON *json, *jval;
         rd_http_error_t *herr;
         rd_bool_t empty;
+        rd_kafka_t *rk;
 
         if (!base_url || !*base_url)
                 RD_UT_SKIP("RD_UT_HTTP_URL environment variable not set");
 
         RD_UT_BEGIN();
 
+        rk             = rd_calloc(1, sizeof(*rk));
         error_url_size = strlen(base_url) + strlen("/error") + 1;
         error_url      = rd_alloca(error_url_size);
         rd_snprintf(error_url, error_url_size, "%s/error", base_url);
 
         /* Try the base url first, parse its JSON and extract a key-value. */
         json = NULL;
-        herr = rd_http_get_json(base_url, &json);
+        herr = rd_http_get_json(rk, base_url, &json);
         RD_UT_ASSERT(!herr, "Expected get_json(%s) to succeed, got: %s",
                      base_url, herr->errstr);
 
@@ -500,7 +647,7 @@ int unittest_http(void) {
 
         /* Try the error URL, verify error code. */
         json = NULL;
-        herr = rd_http_get_json(error_url, &json);
+        herr = rd_http_get_json(rk, error_url, &json);
         RD_UT_ASSERT(herr != NULL, "Expected get_json(%s) to fail", error_url);
         RD_UT_ASSERT(herr->code >= 400,
                      "Expected get_json(%s) error code >= "
@@ -514,6 +661,7 @@ int unittest_http(void) {
         if (json)
                 cJSON_Delete(json);
         rd_http_error_destroy(herr);
+        rd_free(rk);
 
         RD_UT_PASS();
 }

src/rdhttp.h

@@ -42,8 +42,9 @@ typedef struct rd_http_error_s {
 
 void rd_http_error_destroy(rd_http_error_t *herr);
 
-rd_http_error_t *rd_http_get(const char *url, rd_buf_t **rbufp);
-rd_http_error_t *rd_http_get_json(const char *url, cJSON **jsonp);
+rd_http_error_t *rd_http_get(rd_kafka_t *rk, const char *url, rd_buf_t **rbufp);
+rd_http_error_t *
+rd_http_get_json(rd_kafka_t *rk, const char *url, cJSON **jsonp);
 
 void rd_http_global_init(void);
 
@@ -62,7 +63,8 @@ typedef struct rd_http_req_s {
                                                  *   write to. */
 } rd_http_req_t;
 
-rd_http_error_t *rd_http_req_init(rd_http_req_t *hreq, const char *url);
+rd_http_error_t *
+rd_http_req_init(rd_kafka_t *rk, rd_http_req_t *hreq, const char *url);
 rd_http_error_t *rd_http_req_perform_sync(rd_http_req_t *hreq);
 rd_http_error_t *rd_http_parse_json(rd_http_req_t *hreq, cJSON **jsonp);
 rd_http_error_t *rd_http_post_expect_json(rd_kafka_t *rk,

src/rdkafka.c

@@ -5213,13 +5213,8 @@ const char *rd_kafka_get_debug_contexts(void) {
 
 
 int rd_kafka_path_is_dir(const char *path) {
-#ifdef _WIN32
-        struct _stat st;
-        return (_stat(path, &st) == 0 && st.st_mode & S_IFDIR);
-#else
-        struct stat st;
-        return (stat(path, &st) == 0 && S_ISDIR(st.st_mode));
-#endif
+        rd_bool_t is_dir;
+        return rd_file_stat(path, &is_dir) && is_dir;
 }