[KIP-1139] Add support for OAuth jwt-bearer grant type (#4978)

support for `jwt-bearer` grant type as mandated by the KIP.


---------

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

Author: anchitj

CHANGELOG.md

@@ -3,6 +3,7 @@
 librdkafka v2.11.0 is a feature release:
 
 * [KIP-1102](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1102%3A+Enable+clients+to+rebootstrap+based+on+timeout+or+error+code) Enable clients to rebootstrap based on timeout or error code (#4981).
+* [KIP-1139](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1139%3A+Add+support+for+OAuth+jwt-bearer+grant+type) Add support for OAuth jwt-bearer grant type (#4978).
 * Fix for poll ratio calculation in case the queues are forwarded (#5017).
 * Fix data race when buffer queues are being reset instead of being
   initialized (#4718).

CONFIGURATION.md

@@ -100,10 +100,25 @@ enable.sasl.oauthbearer.unsecure.jwt     |  *  | true, false     |         false
 oauthbearer_token_refresh_cb             |  *  |                 |               | low        | SASL/OAUTHBEARER token refresh callback (set with rd_kafka_conf_set_oauthbearer_token_refresh_cb(), triggered by rd_kafka_poll(), et.al. This callback will be triggered when it is time to refresh the client's OAUTHBEARER token. Also see `rd_kafka_conf_enable_sasl_queue()`. <br>*Type: see dedicated API*
 sasl.oauthbearer.method                  |  *  | default, oidc   |       default | low        | Set to "default" or "oidc" to control which login method to be used. If set to "oidc", the following properties must also be be specified: `sasl.oauthbearer.client.id`, `sasl.oauthbearer.client.secret`, and `sasl.oauthbearer.token.endpoint.url`. <br>*Type: enum value*
 sasl.oauthbearer.client.id               |  *  |                 |               | low        | Public identifier for the application. Must be unique across all clients that the authorization server handles. Only used when `sasl.oauthbearer.method` is set to "oidc". <br>*Type: string*
+sasl.oauthbearer.client.credentials.client.id |  *  |                 |               | low        | Alias for `sasl.oauthbearer.client.id`: Public identifier for the application. Must be unique across all clients that the authorization server handles. Only used when `sasl.oauthbearer.method` is set to "oidc". <br>*Type: string*
+sasl.oauthbearer.client.credentials.client.secret |  *  |                 |               | low        | Alias for `sasl.oauthbearer.client.secret`: Client secret only known to the application and the authorization server. This should be a sufficiently random string that is not guessable. Only used when `sasl.oauthbearer.method` is set to "oidc". <br>*Type: string*
 sasl.oauthbearer.client.secret           |  *  |                 |               | low        | Client secret only known to the application and the authorization server. This should be a sufficiently random string that is not guessable. Only used when `sasl.oauthbearer.method` is set to "oidc". <br>*Type: string*
 sasl.oauthbearer.scope                   |  *  |                 |               | low        | Client use this to specify the scope of the access request to the broker. Only used when `sasl.oauthbearer.method` is set to "oidc". <br>*Type: string*
 sasl.oauthbearer.extensions              |  *  |                 |               | low        | Allow additional information to be provided to the broker. Comma-separated list of key=value pairs. E.g., "supportFeatureX=true,organizationId=sales-emea".Only used when `sasl.oauthbearer.method` is set to "oidc". <br>*Type: string*
 sasl.oauthbearer.token.endpoint.url      |  *  |                 |               | low        | OAuth/OIDC issuer token endpoint HTTP(S) URI used to retrieve token. Only used when `sasl.oauthbearer.method` is set to "oidc". <br>*Type: string*
+sasl.oauthbearer.grant.type              |  *  | client_credentials, urn:ietf:params:oauth:grant-type:jwt-bearer | client_credentials | low        | OAuth grant type to use when communicating with the identity provider. <br>*Type: enum value*
+sasl.oauthbearer.assertion.algorithm     |  *  | RS256, ES256    |         RS256 | low        | Algorithm the client should use to sign the assertion sent to the identity provider and in the OAuth alg header in the JWT assertion. <br>*Type: enum value*
+sasl.oauthbearer.assertion.private.key.file |  *  |                 |               | low        | Path to client's private key (PEM) used for authentication when using the JWT assertion. <br>*Type: string*
+sasl.oauthbearer.assertion.private.key.passphrase |  *  |                 |               | low        | Private key passphrase for `sasl.oauthbearer.assertion.private.key.file` or `sasl.oauthbearer.assertion.private.key.pem`. <br>*Type: string*
+sasl.oauthbearer.assertion.private.key.pem |  *  |                 |               | low        | Client's private key (PEM) used for authentication when using the JWT assertion. <br>*Type: string*
+sasl.oauthbearer.assertion.file          |  *  |                 |               | low        | Path to the assertion file. Only used when `sasl.oauthbearer.method` is set to "oidc" and JWT assertion is needed. <br>*Type: string*
+sasl.oauthbearer.assertion.claim.aud     |  *  |                 |               | low        | JWT audience claim. Only used when `sasl.oauthbearer.method` is set to "oidc" and JWT assertion is needed. <br>*Type: string*
+sasl.oauthbearer.assertion.claim.exp.seconds |  *  | 1 .. 2147483647 |           300 | low        | Assertion expiration time in seconds. Only used when `sasl.oauthbearer.method` is set to "oidc" and JWT assertion is needed. <br>*Type: integer*
+sasl.oauthbearer.assertion.claim.iss     |  *  |                 |               | low        | JWT issuer claim. Only used when `sasl.oauthbearer.method` is set to "oidc" and JWT assertion is needed. <br>*Type: string*
+sasl.oauthbearer.assertion.claim.jti.include |  *  | true, false     |         false | low        | JWT ID claim. When set to `true`a random UUID is generated. Only used when `sasl.oauthbearer.method` is set to "oidc" and JWT assertion is needed. <br>*Type: boolean*
+sasl.oauthbearer.assertion.claim.nbf.seconds |  *  | 0 .. 2147483647 |            60 | low        | Assertion not before time in seconds. Only used when `sasl.oauthbearer.method` is set to "oidc" and JWT assertion is needed. <br>*Type: integer*
+sasl.oauthbearer.assertion.claim.sub     |  *  |                 |               | low        | JWT subject claim. Only used when `sasl.oauthbearer.method` is set to "oidc" and JWT assertion is needed. <br>*Type: string*
+sasl.oauthbearer.assertion.jwt.template.file |  *  |                 |               | low        | Path to the JWT template file. Only used when `sasl.oauthbearer.method` is set to "oidc" and JWT assertion is needed. <br>*Type: string*
 plugin.library.paths                     |  *  |                 |               | low        | List of plugin libraries to load (; separated). The library search path is platform dependent (see dlopen(3) for Unix and LoadLibrary() for Windows). If no filename extension is specified the platform-specific extension (such as .dll or .so) will be appended automatically. <br>*Type: string*
 interceptors                             |  *  |                 |               | low        | Interceptors added through rd_kafka_conf_interceptor_add_..() and any configuration handled by interceptors. <br>*Type: see dedicated API*
 group.id                                 |  C  |                 |               | high       | Client group id string. All clients sharing the same group.id belong to the same group. <br>*Type: string*

INTRODUCTION.md

@@ -2137,6 +2137,7 @@ The [Apache Kafka Implementation Proposals (KIPs)](https://cwiki.apache.org/conf
 | KIP-951 - Leader discovery optimisations for the client                  | 3.7.0                       | Supported                                                                                     |
 | KIP-1082 - Require Client-Generated IDs over the ConsumerGroupHeartbeat  | 4.0.0                       | Supported                                                                                     |
 | KIP-1102 - Enable clients to rebootstrap based on timeout or error code  | 4.0.0                       | Supported                                                                                     |
+| KIP-1139 - Add support for OAuth jwt-bearer grant type                   | 4.1.0 (WIP)                 | Supported                                                                                     |
 
 
 

src/rd.h

@@ -438,4 +438,110 @@ typedef struct rd_chariov_s {
         size_t size;
 } rd_chariov_t;
 
+/**
+ * @brief Read the file at \p file_path in binary mode and return its contents.
+ *        The returned buffer is NULL-terminated,
+ *        the size parameter will contain the actual file size.
+ *
+ * @param file_path Path to the file to read.
+ * @param size Pointer to store the file size (optional).
+ * @param max_size Maximum file size to read (0 for no limit) (optional).
+ *
+ * @returns Newly allocated buffer containing the file contents.
+ *          NULL on error (file not found, too large, etc).
+ *
+ * @remark The returned pointer ownership is transferred to the caller.
+ *
+ * @locality Any thread
+ */
+static RD_INLINE RD_UNUSED char *
+rd_file_read(const char *file_path, size_t *size, size_t max_size) {
+        FILE *file;
+        char *buf = NULL;
+        size_t file_size;
+        size_t read_size;
+        if (!size)
+                size = &read_size;
+
+#ifndef _WIN32
+        file = fopen(file_path, "rb");
+#else
+        file  = NULL;
+        errno = fopen_s(&file, file_path, "rb");
+#endif
+        if (!file)
+                return NULL;
+
+        if (fseek(file, 0, SEEK_END) != 0)
+                goto err;
+
+        file_size = (size_t)ftell(file);
+        if (file_size < 0)
+                goto err;
+
+        if (fseek(file, 0, SEEK_SET) != 0)
+                goto err;
+
+        /* Check if file is too large */
+        if (max_size > 0 && file_size > max_size)
+                goto err;
+
+        /* Allocate buffer with extra byte for NULL terminator */
+        buf       = (char *)rd_malloc(file_size + 1);
+        read_size = fread(buf, 1, file_size, file);
+
+        if (read_size != file_size)
+                goto err;
+
+        /* NULL terminate the buffer */
+        buf[file_size] = '\0';
+        *size          = file_size;
+        fclose(file);
+        return buf;
+err:
+        fclose(file);
+        if (buf)
+                rd_free(buf);
+        return NULL;
+}
+
+static RD_INLINE RD_UNUSED FILE *
+rd_file_mkstemp(const char *prefix,
+                const char *mode,
+                char *tempfile_path_out,
+                size_t tempfile_path_out_size) {
+        FILE *tempfile;
+
+#ifdef _WIN32
+        char tempfolder_path[MAX_PATH];
+        char tempfile_path[MAX_PATH];
+        if (!GetTempPathA(MAX_PATH, tempfolder_path))
+                return NULL; /* Failed to get temp folder path */
+
+
+        if (!GetTempFileNameA(tempfolder_path, "TMP", 1, tempfile_path))
+                return NULL; /* Failed to create temp file name */
+
+        tempfile = fopen(tempfile_path, mode);
+#else
+        int tempfile_fd;
+        char tempfile_path[512];
+        rd_snprintf(tempfile_path, sizeof(tempfile_path), "/tmp/%sXXXXXX",
+                    prefix);
+        tempfile_fd = mkstemp(tempfile_path);
+        if (tempfile_fd < 0)
+                return NULL;
+
+        tempfile = fdopen(tempfile_fd, mode);
+#endif
+
+        if (!tempfile)
+                return NULL;
+
+        if (tempfile_path_out)
+                rd_snprintf(tempfile_path_out, tempfile_path_out_size, "%s",
+                            tempfile_path);
+        return tempfile;
+}
+
 #endif /* _RD_H_ */

src/rdbase64.c

@@ -119,6 +119,37 @@ char *rd_base64_encode_str(const rd_chariov_t *in) {
         return out.ptr;
 }
 
+/**
+ * @brief Base64 encode binary input \p in and return a newly allocated,
+ *        base64-encoded string with URL-safe characters.
+ * @returns a newly allocated, base64-encoded string or NULL in case of some
+ *          issue with the conversion or the conversion is not supported.
+ *
+ * @remark Returned string must be freed after use.
+ */
+char *rd_base64_encode_str_urlsafe(const rd_chariov_t *in) {
+        rd_chariov_t out;
+        char *p;
+        rd_base64_encode(in, &out);
+
+        /* Replace + with - and / with _ */
+        for (p = out.ptr; *p; p++) {
+                if (*p == '+')
+                        *p = '-';
+                else if (*p == '/')
+                        *p = '_';
+        }
+
+        /* Remove padding '=' characters */
+        int newlen = strlen(out.ptr);
+        while (newlen > 0 && out.ptr[newlen - 1] == '=') {
+                out.ptr[newlen - 1] = '\0';
+                newlen--;
+        }
+
+        out.size = newlen;
+        return out.ptr;
+}
 
 /**
  * @brief Base64 decode input string \p in. Ignores leading and trailing
@@ -166,4 +197,4 @@ int rd_base64_decode(const rd_chariov_t *in, rd_chariov_t *out) {
 #else
         return -2;
 #endif
-}
+}

src/rdbase64.h

@@ -36,6 +36,8 @@ void rd_base64_encode(const rd_chariov_t *in, rd_chariov_t *out);
 
 char *rd_base64_encode_str(const rd_chariov_t *in);
 
+char *rd_base64_encode_str_urlsafe(const rd_chariov_t *in);
+
 int rd_base64_decode(const rd_chariov_t *in, rd_chariov_t *out);
 
 #endif /* _RDBASE64_H_ */

src/rdkafka.c

@@ -2431,11 +2431,22 @@ rd_kafka_t *rd_kafka_new(rd_kafka_type_t type,
 #if WITH_OAUTHBEARER_OIDC
         if (rk->rk_conf.sasl.oauthbearer.method ==
                 RD_KAFKA_SASL_OAUTHBEARER_METHOD_OIDC &&
-            !rk->rk_conf.sasl.oauthbearer.token_refresh_cb)
-                rd_kafka_conf_set_oauthbearer_token_refresh_cb(
-                    &rk->rk_conf, rd_kafka_oidc_token_refresh_cb);
+            !rk->rk_conf.sasl.oauthbearer.token_refresh_cb) {
+                /* Use JWT bearer */
+                if (rk->rk_conf.sasl.oauthbearer.grant_type ==
+                    RD_KAFKA_SASL_OAUTHBEARER_GRANT_TYPE_CLIENT_CREDENTIALS) {
+                        rd_kafka_conf_set_oauthbearer_token_refresh_cb(
+                            &rk->rk_conf,
+                            rd_kafka_oidc_token_client_credentials_refresh_cb);
+                } else {
+                        rd_kafka_conf_set_oauthbearer_token_refresh_cb(
+                            &rk->rk_conf,
+                            rd_kafka_oidc_token_jwt_bearer_refresh_cb);
+                }
+        }
 #endif
 
+
         rk->rk_controllerid = -1;
 
         /* Admin client defaults */
@@ -5360,7 +5371,7 @@ void rd_kafka_Uuid_destroy(rd_kafka_Uuid_t *uuid) {
  *
  * @remark  Must be freed after use.
  */
-const char *rd_kafka_Uuid_str(const rd_kafka_Uuid_t *uuid) {
+char *rd_kafka_Uuid_str(const rd_kafka_Uuid_t *uuid) {
         int i, j;
         unsigned char bytes[16];
         char *ret = rd_calloc(37, sizeof(*ret));