http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl.c ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl.c b/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl.c deleted file mode 100644 index 9ece5cb..0000000 --- a/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl.c +++ /dev/null @@ -1,343 +0,0 @@ -/* - * librdkafka - The Apache Kafka C/C++ library - * - * Copyright (c) 2015 Magnus Edenhill - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * 1. Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * 2. Redistributions in binary form must reproduce the above copyright notice, - * this list of conditions and the following disclaimer in the documentation - * and/or other materials provided with the distribution. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - */ - -#include "rdkafka_int.h" -#include "rdkafka_transport.h" -#include "rdkafka_transport_int.h" -#include "rdkafka_sasl.h" -#include "rdkafka_sasl_int.h" - - - /** - * Send auth message with framing. - * This is a blocking call. - */ -int rd_kafka_sasl_send (rd_kafka_transport_t *rktrans, - const void *payload, int len, - char *errstr, size_t errstr_size) { - rd_buf_t buf; - rd_slice_t slice; - int32_t hdr; - - rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY, "SASL", - "Send SASL frame to broker (%d bytes)", len); - - rd_buf_init(&buf, 1+1, sizeof(hdr)); - - hdr = htobe32(len); - rd_buf_write(&buf, &hdr, sizeof(hdr)); - if (payload) - rd_buf_push(&buf, payload, len, NULL); - - rd_slice_init_full(&slice, &buf); - - /* Simulate blocking behaviour on non-blocking socket.. - * FIXME: This isn't optimal but is highly unlikely to stall since - * the socket buffer will most likely not be exceeded. */ - do { - int r; - - r = (int)rd_kafka_transport_send(rktrans, &slice, - errstr, errstr_size); - if (r == -1) { - rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY, "SASL", - "SASL send failed: %s", errstr); - rd_buf_destroy(&buf); - return -1; - } - - if (rd_slice_remains(&slice) == 0) - break; - - /* Avoid busy-looping */ - rd_usleep(10*1000, NULL); - - } while (1); - - rd_buf_destroy(&buf); - - return 0; -} - - -/** - * @brief Authentication succesful - * - * Transition to next connect state. - */ -void rd_kafka_sasl_auth_done (rd_kafka_transport_t *rktrans) { - /* Authenticated */ - rd_kafka_broker_connect_up(rktrans->rktrans_rkb); -} - - -int rd_kafka_sasl_io_event (rd_kafka_transport_t *rktrans, int events, - char *errstr, size_t errstr_size) { - rd_kafka_buf_t *rkbuf; - int r; - const void *buf; - size_t len; - - if (!(events & POLLIN)) - return 0; - - r = rd_kafka_transport_framed_recv(rktrans, &rkbuf, - errstr, errstr_size); - if (r == -1) { - if (!strcmp(errstr, "Disconnected")) - rd_snprintf(errstr, errstr_size, - "Disconnected: check client %s credentials " - "and broker logs", - rktrans->rktrans_rkb->rkb_rk->rk_conf. - sasl.mechanisms); - return -1; - } else if (r == 0) /* not fully received yet */ - return 0; - - rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY, "SASL", - "Received SASL frame from broker (%"PRIusz" bytes)", - rkbuf ? rkbuf->rkbuf_totlen : 0); - - if (rkbuf) { - rd_slice_init_full(&rkbuf->rkbuf_reader, &rkbuf->rkbuf_buf); - /* Seek past framing header */ - rd_slice_seek(&rkbuf->rkbuf_reader, 4); - len = rd_slice_remains(&rkbuf->rkbuf_reader); - buf = rd_slice_ensure_contig(&rkbuf->rkbuf_reader, len); - } else { - buf = NULL; - len = 0; - } - - r = rktrans->rktrans_rkb->rkb_rk-> - rk_conf.sasl.provider->recv(rktrans, buf, len, - errstr, errstr_size); - rd_kafka_buf_destroy(rkbuf); - - return r; -} - - -/** - * @brief Close SASL session (from transport code) - * @remark May be called on non-SASL transports (no-op) - */ -void rd_kafka_sasl_close (rd_kafka_transport_t *rktrans) { - const struct rd_kafka_sasl_provider *provider = - rktrans->rktrans_rkb->rkb_rk->rk_conf. - sasl.provider; - - if (provider && provider->close) - provider->close(rktrans); -} - - - -/** - * Initialize and start SASL authentication. - * - * Returns 0 on successful init and -1 on error. - * - * Locality: broker thread - */ -int rd_kafka_sasl_client_new (rd_kafka_transport_t *rktrans, - char *errstr, size_t errstr_size) { - int r; - rd_kafka_broker_t *rkb = rktrans->rktrans_rkb; - rd_kafka_t *rk = rkb->rkb_rk; - char *hostname, *t; - const struct rd_kafka_sasl_provider *provider = - rk->rk_conf.sasl.provider; - - /* Verify broker support: - * - RD_KAFKA_FEATURE_SASL_GSSAPI - GSSAPI supported - * - RD_KAFKA_FEATURE_SASL_HANDSHAKE - GSSAPI, PLAIN and possibly - * other mechanisms supported. */ - if (!strcmp(rk->rk_conf.sasl.mechanisms, "GSSAPI")) { - if (!(rkb->rkb_features & RD_KAFKA_FEATURE_SASL_GSSAPI)) { - rd_snprintf(errstr, errstr_size, - "SASL GSSAPI authentication not supported " - "by broker"); - return -1; - } - } else if (!(rkb->rkb_features & RD_KAFKA_FEATURE_SASL_HANDSHAKE)) { - rd_snprintf(errstr, errstr_size, - "SASL Handshake not supported by broker " - "(required by mechanism %s)%s", - rk->rk_conf.sasl.mechanisms, - rk->rk_conf.api_version_request ? "" : - ": try api.version.request=true"); - return -1; - } - - rd_strdupa(&hostname, rktrans->rktrans_rkb->rkb_nodename); - if ((t = strchr(hostname, ':'))) - *t = '\0'; /* remove ":port" */ - - rd_rkb_dbg(rkb, SECURITY, "SASL", - "Initializing SASL client: service name %s, " - "hostname %s, mechanisms %s, provider %s", - rk->rk_conf.sasl.service_name, hostname, - rk->rk_conf.sasl.mechanisms, - provider->name); - - r = provider->client_new(rktrans, hostname, errstr, errstr_size); - if (r != -1) - rd_kafka_transport_poll_set(rktrans, POLLIN); - - return r; -} - - - - - - - -/** - * Per handle SASL term. - * - * Locality: broker thread - */ -void rd_kafka_sasl_broker_term (rd_kafka_broker_t *rkb) { - const struct rd_kafka_sasl_provider *provider = - rkb->rkb_rk->rk_conf.sasl.provider; - if (provider->broker_term) - provider->broker_term(rkb); -} - -/** - * Broker SASL init. - * - * Locality: broker thread - */ -void rd_kafka_sasl_broker_init (rd_kafka_broker_t *rkb) { - const struct rd_kafka_sasl_provider *provider = - rkb->rkb_rk->rk_conf.sasl.provider; - if (provider->broker_init) - provider->broker_init(rkb); -} - - - -/** - * @brief Select SASL provider for configured mechanism (singularis) - * @returns 0 on success or -1 on failure. - */ -int rd_kafka_sasl_select_provider (rd_kafka_t *rk, - char *errstr, size_t errstr_size) { - const struct rd_kafka_sasl_provider *provider = NULL; - - if (!strcmp(rk->rk_conf.sasl.mechanisms, "GSSAPI")) { - /* GSSAPI / Kerberos */ -#ifdef _MSC_VER - provider = &rd_kafka_sasl_win32_provider; -#elif WITH_SASL_CYRUS - provider = &rd_kafka_sasl_cyrus_provider; -#endif - - } else if (!strcmp(rk->rk_conf.sasl.mechanisms, "PLAIN")) { - /* SASL PLAIN */ - provider = &rd_kafka_sasl_plain_provider; - - } else if (!strncmp(rk->rk_conf.sasl.mechanisms, "SCRAM-SHA-", - strlen("SCRAM-SHA-"))) { - /* SASL SCRAM */ -#if WITH_SASL_SCRAM - provider = &rd_kafka_sasl_scram_provider; -#endif - - } else { - /* Unsupported mechanism */ - rd_snprintf(errstr, errstr_size, - "Unsupported SASL mechanism: %s", - rk->rk_conf.sasl.mechanisms); - return -1; - } - - if (!provider) { - rd_snprintf(errstr, errstr_size, - "No provider for SASL mechanism %s" - ": recompile librdkafka with " -#ifndef _MSC_VER - "libsasl2 or " -#endif - "openssl support. " - "Current build options:" - " PLAIN" -#ifdef _MSC_VER - " WindowsSSPI(GSSAPI)" -#endif -#if WITH_SASL_CYRUS - " SASL_CYRUS" -#endif -#if WITH_SASL_SCRAM - " SASL_SCRAM" -#endif - , - rk->rk_conf.sasl.mechanisms); - return -1; - } - - rd_kafka_dbg(rk, SECURITY, "SASL", - "Selected provider %s for SASL mechanism %s", - provider->name, rk->rk_conf.sasl.mechanisms); - - /* Validate SASL config */ - if (provider->conf_validate && - provider->conf_validate(rk, errstr, errstr_size) == -1) - return -1; - - rk->rk_conf.sasl.provider = provider; - - return 0; -} - - - -/** - * Global SASL termination. - */ -void rd_kafka_sasl_global_term (void) { -#if WITH_SASL_CYRUS - rd_kafka_sasl_cyrus_global_term(); -#endif -} - - -/** - * Global SASL init, called once per runtime. - */ -int rd_kafka_sasl_global_init (void) { -#if WITH_SASL_CYRUS - return rd_kafka_sasl_cyrus_global_init(); -#else - return 0; -#endif -} -
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl.h ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl.h b/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl.h deleted file mode 100644 index 496e04e..0000000 --- a/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl.h +++ /dev/null @@ -1,46 +0,0 @@ -/* - * librdkafka - The Apache Kafka C/C++ library - * - * Copyright (c) 2015 Magnus Edenhill - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * 1. Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * 2. Redistributions in binary form must reproduce the above copyright notice, - * this list of conditions and the following disclaimer in the documentation - * and/or other materials provided with the distribution. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - */ - -#pragma once - - - -int rd_kafka_sasl_io_event (rd_kafka_transport_t *rktrans, int events, - char *errstr, size_t errstr_size); -void rd_kafka_sasl_close (rd_kafka_transport_t *rktrans); -int rd_kafka_sasl_client_new (rd_kafka_transport_t *rktrans, - char *errstr, size_t errstr_size); - -void rd_kafka_sasl_broker_term (rd_kafka_broker_t *rkb); -void rd_kafka_sasl_broker_init (rd_kafka_broker_t *rkb); - -void rd_kafka_sasl_global_term (void); -int rd_kafka_sasl_global_init (void); - -int rd_kafka_sasl_select_provider (rd_kafka_t *rk, - char *errstr, size_t errstr_size); http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl_cyrus.c ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl_cyrus.c b/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl_cyrus.c deleted file mode 100644 index 35b3183..0000000 --- a/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl_cyrus.c +++ /dev/null @@ -1,623 +0,0 @@ -/* - * librdkafka - The Apache Kafka C/C++ library - * - * Copyright (c) 2015 Magnus Edenhill - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * 1. Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * 2. Redistributions in binary form must reproduce the above copyright notice, - * this list of conditions and the following disclaimer in the documentation - * and/or other materials provided with the distribution. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - */ - -#include "rdkafka_int.h" -#include "rdkafka_transport.h" -#include "rdkafka_transport_int.h" -#include "rdkafka_sasl.h" -#include "rdkafka_sasl_int.h" -#include "rdstring.h" - -#ifdef __FreeBSD__ -#include <sys/wait.h> /* For WIF.. */ -#endif - -#ifdef __APPLE__ -/* Apple has deprecated most of the SASL API for unknown reason, - * silence those warnings. */ -#pragma clang diagnostic ignored "-Wdeprecated-declarations" -#endif - -#include <sasl/sasl.h> - -static mtx_t rd_kafka_sasl_cyrus_kinit_lock; - -typedef struct rd_kafka_sasl_cyrus_state_s { - sasl_conn_t *conn; - sasl_callback_t callbacks[16]; -} rd_kafka_sasl_cyrus_state_t; - - - -/** - * Handle received frame from broker. - */ -static int rd_kafka_sasl_cyrus_recv (struct rd_kafka_transport_s *rktrans, - const void *buf, size_t size, - char *errstr, size_t errstr_size) { - rd_kafka_sasl_cyrus_state_t *state = rktrans->rktrans_sasl.state; - int r; - - if (rktrans->rktrans_sasl.complete && size == 0) - goto auth_successful; - - do { - sasl_interact_t *interact = NULL; - const char *out; - unsigned int outlen; - - r = sasl_client_step(state->conn, - size > 0 ? buf : NULL, size, - &interact, - &out, &outlen); - - if (r >= 0) { - /* Note: outlen may be 0 here for an empty response */ - if (rd_kafka_sasl_send(rktrans, out, outlen, - errstr, errstr_size) == -1) - return -1; - } - - if (r == SASL_INTERACT) - rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY, "SASL", - "SASL_INTERACT: %lu %s, %s, %s, %p", - interact->id, - interact->challenge, - interact->prompt, - interact->defresult, - interact->result); - - } while (r == SASL_INTERACT); - - if (r == SASL_CONTINUE) - return 0; /* Wait for more data from broker */ - else if (r != SASL_OK) { - rd_snprintf(errstr, errstr_size, - "SASL handshake failed (step): %s", - sasl_errdetail(state->conn)); - return -1; - } - - /* Authentication successful */ -auth_successful: - if (rktrans->rktrans_rkb->rkb_rk->rk_conf.debug & - RD_KAFKA_DBG_SECURITY) { - const char *user, *mech, *authsrc; - - if (sasl_getprop(state->conn, SASL_USERNAME, - (const void **)&user) != SASL_OK) - user = "(unknown)"; - - if (sasl_getprop(state->conn, SASL_MECHNAME, - (const void **)&mech) != SASL_OK) - mech = "(unknown)"; - - if (sasl_getprop(state->conn, SASL_AUTHSOURCE, - (const void **)&authsrc) != SASL_OK) - authsrc = "(unknown)"; - - rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY, "SASL", - "Authenticated as %s using %s (%s)", - user, mech, authsrc); - } - - rd_kafka_sasl_auth_done(rktrans); - - return 0; -} - - - - -static ssize_t render_callback (const char *key, char *buf, - size_t size, void *opaque) { - rd_kafka_broker_t *rkb = opaque; - - if (!strcmp(key, "broker.name")) { - char *val, *t; - size_t len; - rd_kafka_broker_lock(rkb); - rd_strdupa(&val, rkb->rkb_nodename); - rd_kafka_broker_unlock(rkb); - - /* Just the broker name, no port */ - if ((t = strchr(val, ':'))) - len = (size_t)(t-val); - else - len = strlen(val); - - if (buf) - memcpy(buf, val, RD_MIN(len, size)); - - return len; - - } else { - rd_kafka_conf_res_t res; - size_t destsize = size; - - /* Try config lookup. */ - res = rd_kafka_conf_get(&rkb->rkb_rk->rk_conf, key, - buf, &destsize); - if (res != RD_KAFKA_CONF_OK) - return -1; - - /* Dont include \0 in returned size */ - return (destsize > 0 ? destsize-1 : destsize); - } -} - - -/** - * Execute kinit to refresh ticket. - * - * Returns 0 on success, -1 on error. - * - * Locality: any - */ -static int rd_kafka_sasl_cyrus_kinit_refresh (rd_kafka_broker_t *rkb) { - rd_kafka_t *rk = rkb->rkb_rk; - int r; - char *cmd; - char errstr[128]; - - if (!rk->rk_conf.sasl.kinit_cmd || - !strstr(rk->rk_conf.sasl.mechanisms, "GSSAPI")) - return 0; /* kinit not configured */ - - /* Build kinit refresh command line using string rendering and config */ - cmd = rd_string_render(rk->rk_conf.sasl.kinit_cmd, - errstr, sizeof(errstr), - render_callback, rkb); - if (!cmd) { - rd_rkb_log(rkb, LOG_ERR, "SASLREFRESH", - "Failed to construct kinit command " - "from sasl.kerberos.kinit.cmd template: %s", - errstr); - return -1; - } - - /* Execute kinit */ - rd_rkb_dbg(rkb, SECURITY, "SASLREFRESH", - "Refreshing SASL keys with command: %s", cmd); - - mtx_lock(&rd_kafka_sasl_cyrus_kinit_lock); - r = system(cmd); - mtx_unlock(&rd_kafka_sasl_cyrus_kinit_lock); - - if (r == -1) { - rd_rkb_log(rkb, LOG_ERR, "SASLREFRESH", - "SASL key refresh failed: Failed to execute %s", - cmd); - rd_free(cmd); - return -1; - } else if (WIFSIGNALED(r)) { - rd_rkb_log(rkb, LOG_ERR, "SASLREFRESH", - "SASL key refresh failed: %s: received signal %d", - cmd, WTERMSIG(r)); - rd_free(cmd); - return -1; - } else if (WIFEXITED(r) && WEXITSTATUS(r) != 0) { - rd_rkb_log(rkb, LOG_ERR, "SASLREFRESH", - "SASL key refresh failed: %s: exited with code %d", - cmd, WEXITSTATUS(r)); - rd_free(cmd); - return -1; - } - - rd_free(cmd); - - rd_rkb_dbg(rkb, SECURITY, "SASLREFRESH", "SASL key refreshed"); - return 0; -} - - -/** - * Refresh timer callback - * - * Locality: kafka main thread - */ -static void rd_kafka_sasl_cyrus_kinit_refresh_tmr_cb (rd_kafka_timers_t *rkts, - void *arg) { - rd_kafka_broker_t *rkb = arg; - - rd_kafka_sasl_cyrus_kinit_refresh(rkb); -} - - - -/** - * - * libsasl callbacks - * - */ -static RD_UNUSED int -rd_kafka_sasl_cyrus_cb_getopt (void *context, const char *plugin_name, - const char *option, - const char **result, unsigned *len) { - rd_kafka_transport_t *rktrans = context; - - if (!strcmp(option, "client_mech_list")) - *result = "GSSAPI"; - if (!strcmp(option, "canon_user_plugin")) - *result = "INTERNAL"; - - if (*result && len) - *len = strlen(*result); - - rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY, "LIBSASL", - "CB_GETOPT: plugin %s, option %s: returning %s", - plugin_name, option, *result); - - return SASL_OK; -} - -static int rd_kafka_sasl_cyrus_cb_log (void *context, int level, const char *message){ - rd_kafka_transport_t *rktrans = context; - - if (level >= LOG_DEBUG) - rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY, "LIBSASL", - "%s", message); - else - rd_rkb_log(rktrans->rktrans_rkb, level, "LIBSASL", - "%s", message); - return SASL_OK; -} - - -static int rd_kafka_sasl_cyrus_cb_getsimple (void *context, int id, - const char **result, unsigned *len) { - rd_kafka_transport_t *rktrans = context; - - switch (id) - { - case SASL_CB_USER: - case SASL_CB_AUTHNAME: - *result = rktrans->rktrans_rkb->rkb_rk->rk_conf.sasl.username; - break; - - default: - *result = NULL; - break; - } - - if (len) - *len = *result ? strlen(*result) : 0; - - rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY, "LIBSASL", - "CB_GETSIMPLE: id 0x%x: returning %s", id, *result); - - return *result ? SASL_OK : SASL_FAIL; -} - - -static int rd_kafka_sasl_cyrus_cb_getsecret (sasl_conn_t *conn, void *context, - int id, sasl_secret_t **psecret) { - rd_kafka_transport_t *rktrans = context; - const char *password; - - password = rktrans->rktrans_rkb->rkb_rk->rk_conf.sasl.password; - - if (!password) { - *psecret = NULL; - } else { - size_t passlen = strlen(password); - *psecret = rd_realloc(*psecret, sizeof(**psecret) + passlen); - (*psecret)->len = passlen; - memcpy((*psecret)->data, password, passlen); - } - - rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY, "LIBSASL", - "CB_GETSECRET: id 0x%x: returning %s", - id, *psecret ? "(hidden)":"NULL"); - - return SASL_OK; -} - -static int rd_kafka_sasl_cyrus_cb_chalprompt (void *context, int id, - const char *challenge, - const char *prompt, - const char *defres, - const char **result, unsigned *len) { - rd_kafka_transport_t *rktrans = context; - - *result = "min_chalprompt"; - *len = strlen(*result); - - rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY, "LIBSASL", - "CB_CHALPROMPT: id 0x%x, challenge %s, prompt %s, " - "default %s: returning %s", - id, challenge, prompt, defres, *result); - - return SASL_OK; -} - -static int rd_kafka_sasl_cyrus_cb_getrealm (void *context, int id, - const char **availrealms, - const char **result) { - rd_kafka_transport_t *rktrans = context; - - *result = *availrealms; - - rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY, "LIBSASL", - "CB_GETREALM: id 0x%x: returning %s", id, *result); - - return SASL_OK; -} - - -static RD_UNUSED int -rd_kafka_sasl_cyrus_cb_canon (sasl_conn_t *conn, - void *context, - const char *in, unsigned inlen, - unsigned flags, - const char *user_realm, - char *out, unsigned out_max, - unsigned *out_len) { - rd_kafka_transport_t *rktrans = context; - - if (strstr(rktrans->rktrans_rkb->rkb_rk->rk_conf. - sasl.mechanisms, "GSSAPI")) { - *out_len = rd_snprintf(out, out_max, "%s", - rktrans->rktrans_rkb->rkb_rk-> - rk_conf.sasl.principal); - } else if (!strcmp(rktrans->rktrans_rkb->rkb_rk->rk_conf. - sasl.mechanisms, "PLAIN")) { - *out_len = rd_snprintf(out, out_max, "%.*s", inlen, in); - } else - out = NULL; - - rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY, "LIBSASL", - "CB_CANON: flags 0x%x, \"%.*s\" @ \"%s\": returning \"%.*s\"", - flags, (int)inlen, in, user_realm, (int)(*out_len), out); - - return out ? SASL_OK : SASL_FAIL; -} - - -static void rd_kafka_sasl_cyrus_close (struct rd_kafka_transport_s *rktrans) { - rd_kafka_sasl_cyrus_state_t *state = rktrans->rktrans_sasl.state; - - if (!state) - return; - - if (state->conn) - sasl_dispose(&state->conn); - rd_free(state); -} - - -/** - * Initialize and start SASL authentication. - * - * Returns 0 on successful init and -1 on error. - * - * Locality: broker thread - */ -static int rd_kafka_sasl_cyrus_client_new (rd_kafka_transport_t *rktrans, - const char *hostname, - char *errstr, size_t errstr_size) { - int r; - rd_kafka_sasl_cyrus_state_t *state; - rd_kafka_broker_t *rkb = rktrans->rktrans_rkb; - rd_kafka_t *rk = rkb->rkb_rk; - sasl_callback_t callbacks[16] = { - // { SASL_CB_GETOPT, (void *)rd_kafka_sasl_cyrus_cb_getopt, rktrans }, - { SASL_CB_LOG, (void *)rd_kafka_sasl_cyrus_cb_log, rktrans }, - { SASL_CB_AUTHNAME, (void *)rd_kafka_sasl_cyrus_cb_getsimple, rktrans }, - { SASL_CB_PASS, (void *)rd_kafka_sasl_cyrus_cb_getsecret, rktrans }, - { SASL_CB_ECHOPROMPT, (void *)rd_kafka_sasl_cyrus_cb_chalprompt, rktrans }, - { SASL_CB_GETREALM, (void *)rd_kafka_sasl_cyrus_cb_getrealm, rktrans }, - { SASL_CB_CANON_USER, (void *)rd_kafka_sasl_cyrus_cb_canon, rktrans }, - { SASL_CB_LIST_END } - }; - - state = rd_calloc(1, sizeof(*state)); - rktrans->rktrans_sasl.state = state; - - /* SASL_CB_USER is needed for PLAIN but breaks GSSAPI */ - if (!strcmp(rk->rk_conf.sasl.mechanisms, "PLAIN")) { - int endidx; - /* Find end of callbacks array */ - for (endidx = 0 ; - callbacks[endidx].id != SASL_CB_LIST_END ; endidx++) - ; - - callbacks[endidx].id = SASL_CB_USER; - callbacks[endidx].proc = (void *)rd_kafka_sasl_cyrus_cb_getsimple; - callbacks[endidx].context = rktrans; - endidx++; - callbacks[endidx].id = SASL_CB_LIST_END; - } - - memcpy(state->callbacks, callbacks, sizeof(callbacks)); - - /* Acquire or refresh ticket if kinit is configured */ - rd_kafka_sasl_cyrus_kinit_refresh(rkb); - - r = sasl_client_new(rk->rk_conf.sasl.service_name, hostname, - NULL, NULL, /* no local & remote IP checks */ - state->callbacks, 0, &state->conn); - if (r != SASL_OK) { - rd_snprintf(errstr, errstr_size, "%s", - sasl_errstring(r, NULL, NULL)); - return -1; - } - - if (rk->rk_conf.debug & RD_KAFKA_DBG_SECURITY) { - const char *avail_mechs; - sasl_listmech(state->conn, NULL, NULL, " ", NULL, - &avail_mechs, NULL, NULL); - rd_rkb_dbg(rkb, SECURITY, "SASL", - "My supported SASL mechanisms: %s", avail_mechs); - } - - do { - const char *out; - unsigned int outlen; - const char *mech = NULL; - - r = sasl_client_start(state->conn, - rk->rk_conf.sasl.mechanisms, - NULL, &out, &outlen, &mech); - - if (r >= 0) - if (rd_kafka_sasl_send(rktrans, out, outlen, - errstr, errstr_size)) - return -1; - } while (r == SASL_INTERACT); - - if (r == SASL_OK) { - /* PLAIN is appearantly done here, but we still need to make sure - * the PLAIN frame is sent and we get a response back (but we must - * not pass the response to libsasl or it will fail). */ - rktrans->rktrans_sasl.complete = 1; - return 0; - - } else if (r != SASL_CONTINUE) { - rd_snprintf(errstr, errstr_size, - "SASL handshake failed (start (%d)): %s", - r, sasl_errdetail(state->conn)); - return -1; - } - - return 0; -} - - - - - - - -/** - * Per handle SASL term. - * - * Locality: broker thread - */ -static void rd_kafka_sasl_cyrus_broker_term (rd_kafka_broker_t *rkb) { - rd_kafka_t *rk = rkb->rkb_rk; - - if (!rk->rk_conf.sasl.kinit_cmd) - return; - - rd_kafka_timer_stop(&rk->rk_timers, &rkb->rkb_sasl_kinit_refresh_tmr,1); -} - -/** - * Broker SASL init. - * - * Locality: broker thread - */ -static void rd_kafka_sasl_cyrus_broker_init (rd_kafka_broker_t *rkb) { - rd_kafka_t *rk = rkb->rkb_rk; - - if (!rk->rk_conf.sasl.kinit_cmd || - !strstr(rk->rk_conf.sasl.mechanisms, "GSSAPI")) - return; /* kinit not configured, no need to start timer */ - - rd_kafka_timer_start(&rk->rk_timers, &rkb->rkb_sasl_kinit_refresh_tmr, - rk->rk_conf.sasl.relogin_min_time * 1000ll, - rd_kafka_sasl_cyrus_kinit_refresh_tmr_cb, rkb); -} - - - -static int rd_kafka_sasl_cyrus_conf_validate (rd_kafka_t *rk, - char *errstr, size_t errstr_size) { - - if (strcmp(rk->rk_conf.sasl.mechanisms, "GSSAPI")) - return 0; - - if (rk->rk_conf.sasl.kinit_cmd) { - rd_kafka_broker_t rkb; - char *cmd; - char tmperr[128]; - - memset(&rkb, 0, sizeof(rkb)); - strcpy(rkb.rkb_nodename, "ATestBroker:9092"); - rkb.rkb_rk = rk; - mtx_init(&rkb.rkb_lock, mtx_plain); - - cmd = rd_string_render(rk->rk_conf.sasl.kinit_cmd, - tmperr, sizeof(tmperr), - render_callback, &rkb); - - mtx_destroy(&rkb.rkb_lock); - - if (!cmd) { - rd_snprintf(errstr, errstr_size, - "Invalid sasl.kerberos.kinit.cmd value: %s", - tmperr); - return -1; - } - - rd_free(cmd); - } - - return 0; -} - - -/** - * Global SASL termination. - */ -void rd_kafka_sasl_cyrus_global_term (void) { - /* NOTE: Should not be called since the application may be using SASL too*/ - /* sasl_done(); */ - mtx_destroy(&rd_kafka_sasl_cyrus_kinit_lock); -} - - -/** - * Global SASL init, called once per runtime. - */ -int rd_kafka_sasl_cyrus_global_init (void) { - int r; - - mtx_init(&rd_kafka_sasl_cyrus_kinit_lock, mtx_plain); - - r = sasl_client_init(NULL); - if (r != SASL_OK) { - fprintf(stderr, "librdkafka: sasl_client_init() failed: %s\n", - sasl_errstring(r, NULL, NULL)); - return -1; - } - - return 0; -} - - -const struct rd_kafka_sasl_provider rd_kafka_sasl_cyrus_provider = { - .name = "Cyrus", - .client_new = rd_kafka_sasl_cyrus_client_new, - .recv = rd_kafka_sasl_cyrus_recv, - .close = rd_kafka_sasl_cyrus_close, - .broker_init = rd_kafka_sasl_cyrus_broker_init, - .broker_term = rd_kafka_sasl_cyrus_broker_term, - .conf_validate = rd_kafka_sasl_cyrus_conf_validate -}; http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl_int.h ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl_int.h b/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl_int.h deleted file mode 100644 index 699174e..0000000 --- a/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl_int.h +++ /dev/null @@ -1,70 +0,0 @@ -/* - * librdkafka - The Apache Kafka C/C++ library - * - * Copyright (c) 2015 Magnus Edenhill - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * 1. Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * 2. Redistributions in binary form must reproduce the above copyright notice, - * this list of conditions and the following disclaimer in the documentation - * and/or other materials provided with the distribution. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - */ - -#pragma once - -struct rd_kafka_sasl_provider { - const char *name; - - int (*client_new) (rd_kafka_transport_t *rktrans, - const char *hostname, - char *errstr, size_t errstr_size); - - int (*recv) (struct rd_kafka_transport_s *s, - const void *buf, size_t size, - char *errstr, size_t errstr_size); - void (*close) (struct rd_kafka_transport_s *); - - void (*broker_init) (rd_kafka_broker_t *rkb); - void (*broker_term) (rd_kafka_broker_t *rkb); - - int (*conf_validate) (rd_kafka_t *rk, - char *errstr, size_t errstr_size); -}; - -#ifdef _MSC_VER -extern const struct rd_kafka_sasl_provider rd_kafka_sasl_win32_provider; -#endif - -#if WITH_SASL_CYRUS -extern const struct rd_kafka_sasl_provider rd_kafka_sasl_cyrus_provider; -void rd_kafka_sasl_cyrus_global_term (void); -int rd_kafka_sasl_cyrus_global_init (void); -#endif - -extern const struct rd_kafka_sasl_provider rd_kafka_sasl_plain_provider; - -#if WITH_SASL_SCRAM -extern const struct rd_kafka_sasl_provider rd_kafka_sasl_scram_provider; -#endif - -void rd_kafka_sasl_auth_done (rd_kafka_transport_t *rktrans); -int rd_kafka_sasl_send (rd_kafka_transport_t *rktrans, - const void *payload, int len, - char *errstr, size_t errstr_size); - http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl_plain.c ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl_plain.c b/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl_plain.c deleted file mode 100644 index 57650ee..0000000 --- a/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl_plain.c +++ /dev/null @@ -1,128 +0,0 @@ -/* - * librdkafka - The Apache Kafka C/C++ library - * - * Copyright (c) 2017 Magnus Edenhill - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * 1. Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * 2. Redistributions in binary form must reproduce the above copyright notice, - * this list of conditions and the following disclaimer in the documentation - * and/or other materials provided with the distribution. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - */ - - -/** - * Builtin SASL PLAIN support when Cyrus SASL is not available - */ -#include "rdkafka_int.h" -#include "rdkafka_transport.h" -#include "rdkafka_transport_int.h" -#include "rdkafka_sasl.h" -#include "rdkafka_sasl_int.h" - - -/** - * @brief Handle received frame from broker. - */ -static int rd_kafka_sasl_plain_recv (struct rd_kafka_transport_s *rktrans, - const void *buf, size_t size, - char *errstr, size_t errstr_size) { - if (size) - rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY, "SASLPLAIN", - "Received non-empty SASL PLAIN (builtin) " - "response from broker (%"PRIusz" bytes)", size); - - rd_kafka_sasl_auth_done(rktrans); - - return 0; -} - - -/** - * @brief Initialize and start SASL PLAIN (builtin) authentication. - * - * Returns 0 on successful init and -1 on error. - * - * @locality broker thread - */ -int rd_kafka_sasl_plain_client_new (rd_kafka_transport_t *rktrans, - const char *hostname, - char *errstr, size_t errstr_size) { - rd_kafka_broker_t *rkb = rktrans->rktrans_rkb; - rd_kafka_t *rk = rkb->rkb_rk; - /* [authzid] UTF8NUL authcid UTF8NUL passwd */ - char buf[255+1+255+1+255+1]; - int of = 0; - - /* authzid: none (empty) */ - /* UTF8NUL */ - buf[of++] = 0; - /* authcid */ - if (rk->rk_conf.sasl.username) { - int r = (int)strlen(rk->rk_conf.sasl.username); - r = RD_MIN(255, r); - memcpy(&buf[of], rk->rk_conf.sasl.username, r); - of += r; - } - /* UTF8NUL */ - buf[of++] = 0; - /* passwd */ - if (rk->rk_conf.sasl.password) { - int r = (int)strlen(rk->rk_conf.sasl.password); - r = RD_MIN(255, r); - memcpy(&buf[of], rk->rk_conf.sasl.password, r); - of += r; - } - - rd_rkb_dbg(rkb, SECURITY, "SASLPLAIN", - "Sending SASL PLAIN (builtin) authentication token"); - - if (rd_kafka_sasl_send(rktrans, buf, of, - errstr, errstr_size)) - return -1; - - /* PLAIN is appearantly done here, but we still need to make sure - * the PLAIN frame is sent and we get a response back (empty) */ - rktrans->rktrans_sasl.complete = 1; - return 0; -} - - -/** - * @brief Validate PLAIN config - */ -static int rd_kafka_sasl_plain_conf_validate (rd_kafka_t *rk, - char *errstr, - size_t errstr_size) { - if (!rk->rk_conf.sasl.username || !rk->rk_conf.sasl.password) { - rd_snprintf(errstr, errstr_size, - "sasl.username and sasl.password must be set"); - return -1; - } - - return 0; -} - - -const struct rd_kafka_sasl_provider rd_kafka_sasl_plain_provider = { - .name = "PLAIN (builtin)", - .client_new = rd_kafka_sasl_plain_client_new, - .recv = rd_kafka_sasl_plain_recv, - .conf_validate = rd_kafka_sasl_plain_conf_validate -}; http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl_scram.c ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl_scram.c b/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl_scram.c deleted file mode 100644 index 968d879..0000000 --- a/thirdparty/librdkafka-0.11.1/src/rdkafka_sasl_scram.c +++ /dev/null @@ -1,901 +0,0 @@ -/* - * librdkafka - The Apache Kafka C/C++ library - * - * Copyright (c) 2017 Magnus Edenhill - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * 1. Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * 2. Redistributions in binary form must reproduce the above copyright notice, - * this list of conditions and the following disclaimer in the documentation - * and/or other materials provided with the distribution. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - */ - - -/** - * Builtin SASL SCRAM support when Cyrus SASL is not available - */ -#include "rdkafka_int.h" -#include "rdkafka_transport.h" -#include "rdkafka_transport_int.h" -#include "rdkafka_sasl.h" -#include "rdkafka_sasl_int.h" -#include "rdrand.h" - -#if WITH_SSL -#include <openssl/hmac.h> -#include <openssl/evp.h> -#include <openssl/sha.h> -#else -#error "WITH_SSL (OpenSSL) is required for SASL SCRAM" -#endif - - -/** - * @brief Per-connection state - */ -struct rd_kafka_sasl_scram_state { - enum { - RD_KAFKA_SASL_SCRAM_STATE_CLIENT_FIRST_MESSAGE, - RD_KAFKA_SASL_SCRAM_STATE_SERVER_FIRST_MESSAGE, - RD_KAFKA_SASL_SCRAM_STATE_CLIENT_FINAL_MESSAGE, - } state; - rd_chariov_t cnonce; /* client c-nonce */ - rd_chariov_t first_msg_bare; /* client-first-message-bare */ - char *ServerSignatureB64; /* ServerSignature in Base64 */ - const EVP_MD *evp; /* Hash function pointer */ -}; - - -/** - * @brief Close and free authentication state - */ -static void rd_kafka_sasl_scram_close (rd_kafka_transport_t *rktrans) { - struct rd_kafka_sasl_scram_state *state = rktrans->rktrans_sasl.state; - - if (!state) - return; - - RD_IF_FREE(state->cnonce.ptr, rd_free); - RD_IF_FREE(state->first_msg_bare.ptr, rd_free); - RD_IF_FREE(state->ServerSignatureB64, rd_free); - rd_free(state); -} - - - -/** - * @brief Generates a nonce string (a random printable string) - * @remark dst->ptr will be allocated and must be freed. - */ -static void rd_kafka_sasl_scram_generate_nonce (rd_chariov_t *dst) { - int i; - dst->size = 32; - dst->ptr = rd_malloc(dst->size+1); - for (i = 0 ; i < (int)dst->size ; i++) - dst->ptr[i] = 'a'; // (char)rd_jitter(0x2d/*-*/, 0x7e/*~*/); - dst->ptr[i] = 0; -} - - -/** - * @brief Parses inbuf for SCRAM attribute \p attr (e.g., 's') - * @returns a newly allocated copy of the value, or NULL - * on failure in which case an error is written to \p errstr - * prefixed by \p description. - */ -static char *rd_kafka_sasl_scram_get_attr (const rd_chariov_t *inbuf, char attr, - const char *description, - char *errstr, size_t errstr_size) { - size_t of = 0; - - for (of = 0 ; of < inbuf->size ; ) { - const char *td; - size_t len; - - /* Find next delimiter , (if any) */ - td = memchr(&inbuf->ptr[of], ',', inbuf->size - of); - if (td) - len = (size_t)(td - &inbuf->ptr[of]); - else - len = inbuf->size - of; - - /* Check if attr "x=" matches */ - if (inbuf->ptr[of] == attr && inbuf->size > of+1 && - inbuf->ptr[of+1] == '=') { - char *ret; - of += 2; /* past = */ - ret = rd_malloc(len - 2 + 1); - memcpy(ret, &inbuf->ptr[of], len - 2); - ret[len-2] = '\0'; - return ret; - } - - /* Not the attr we are looking for, skip - * past the next delimiter and continue looking. */ - of += len+1; - } - - rd_snprintf(errstr, errstr_size, - "%s: could not find attribute (%c)", - description, attr); - return NULL; -} - - -/** - * @brief Base64 encode binary input \p in - * @returns a newly allocated base64 string - */ -static char *rd_base64_encode (const rd_chariov_t *in) { - BIO *buf, *b64f; - BUF_MEM *ptr; - char *out; - - b64f = BIO_new(BIO_f_base64()); - buf = BIO_new(BIO_s_mem()); - buf = BIO_push(b64f, buf); - - BIO_set_flags(buf, BIO_FLAGS_BASE64_NO_NL); - BIO_set_close(buf, BIO_CLOSE); - BIO_write(buf, in->ptr, (int)in->size); - BIO_flush(buf); - - BIO_get_mem_ptr(buf, &ptr); - out = malloc(ptr->length + 1); - memcpy(out, ptr->data, ptr->length); - out[ptr->length] = '\0'; - - BIO_free_all(buf); - - return out; -} - -/** - * @brief Base64 decode input string \p in of size \p insize. - * @returns -1 on invalid Base64, or 0 on successes in which case a - * newly allocated binary string is set in out (and size). - */ -static int rd_base64_decode (const rd_chariov_t *in, rd_chariov_t *out) { - size_t asize; - BIO *b64, *bmem; - - if (in->size == 0 || (in->size % 4) != 0) - return -1; - - asize = (in->size * 3) / 4; /* allocation size */ - out->ptr = rd_malloc(asize+1); - - b64 = BIO_new(BIO_f_base64()); - BIO_set_flags(b64, BIO_FLAGS_BASE64_NO_NL); - - bmem = BIO_new_mem_buf(in->ptr, (int)in->size); - bmem = BIO_push(b64, bmem); - - out->size = BIO_read(bmem, out->ptr, (int)asize+1); - assert(out->size <= asize); - BIO_free_all(bmem); - -#if ENABLE_DEVEL - /* Verify that decode==encode */ - { - char *encoded = rd_base64_encode(out); - assert(strlen(encoded) == in->size); - assert(!strncmp(encoded, in->ptr, in->size)); - rd_free(encoded); - } -#endif - - return 0; -} - - -/** - * @brief Perform H(str) hash function and stores the result in \p out - * which must be at least EVP_MAX_MD_SIZE. - * @returns 0 on success, else -1 - */ -static int -rd_kafka_sasl_scram_H (rd_kafka_transport_t *rktrans, - const rd_chariov_t *str, - rd_chariov_t *out) { - - rktrans->rktrans_rkb->rkb_rk->rk_conf.sasl.scram_H( - (const unsigned char *)str->ptr, str->size, - (unsigned char *)out->ptr); - - out->size = rktrans->rktrans_rkb->rkb_rk->rk_conf.sasl.scram_H_size; - return 0; -} - -/** - * @brief Perform HMAC(key,str) and stores the result in \p out - * which must be at least EVP_MAX_MD_SIZE. - * @returns 0 on success, else -1 - */ -static int -rd_kafka_sasl_scram_HMAC (rd_kafka_transport_t *rktrans, - const rd_chariov_t *key, - const rd_chariov_t *str, - rd_chariov_t *out) { - const EVP_MD *evp = - rktrans->rktrans_rkb->rkb_rk->rk_conf.sasl.scram_evp; - unsigned int outsize; - - //printf("HMAC KEY: %s\n", rd_base64_encode(key)); - //printf("HMAC STR: %s\n", rd_base64_encode(str)); - - if (!HMAC(evp, - (const unsigned char *)key->ptr, (int)key->size, - (const unsigned char *)str->ptr, (int)str->size, - (unsigned char *)out->ptr, &outsize)) { - rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY, "SCRAM", - "HMAC failed"); - return -1; - } - - out->size = outsize; - //printf("HMAC OUT: %s\n", rd_base64_encode(out)); - - return 0; -} - - - -/** - * @brief Perform \p itcnt iterations of HMAC() on the given buffer \p in - * using \p salt, writing the output into \p out which must be - * at least EVP_MAX_MD_SIZE. Actual size is updated in \p *outsize. - * @returns 0 on success, else -1 - */ -static int -rd_kafka_sasl_scram_Hi (rd_kafka_transport_t *rktrans, - const rd_chariov_t *in, - const rd_chariov_t *salt, - int itcnt, rd_chariov_t *out) { - const EVP_MD *evp = - rktrans->rktrans_rkb->rkb_rk->rk_conf.sasl.scram_evp; - unsigned int ressize = 0; - unsigned char tempres[EVP_MAX_MD_SIZE]; - unsigned char *saltplus; - int i; - - /* U1 := HMAC(str, salt + INT(1)) */ - saltplus = rd_alloca(salt->size + 4); - memcpy(saltplus, salt->ptr, salt->size); - saltplus[salt->size] = 0; - saltplus[salt->size+1] = 0; - saltplus[salt->size+2] = 0; - saltplus[salt->size+3] = 1; - - /* U1 := HMAC(str, salt + INT(1)) */ - if (!HMAC(evp, - (const unsigned char *)in->ptr, (int)in->size, - saltplus, salt->size+4, - tempres, &ressize)) { - rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY, "SCRAM", - "HMAC priming failed"); - return -1; - } - - memcpy(out->ptr, tempres, ressize); - - /* Ui-1 := HMAC(str, Ui-2) .. */ - for (i = 1 ; i < itcnt ; i++) { - unsigned char tempdest[EVP_MAX_MD_SIZE]; - int j; - - if (unlikely(!HMAC(evp, - (const unsigned char *)in->ptr, (int)in->size, - tempres, ressize, - tempdest, NULL))) { - rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY, "SCRAM", - "Hi() HMAC #%d/%d failed", i, itcnt); - return -1; - } - - /* U1 XOR U2 .. */ - for (j = 0 ; j < (int)ressize ; j++) { - out->ptr[j] ^= tempdest[j]; - tempres[j] = tempdest[j]; - } - } - - out->size = ressize; - - return 0; -} - - -/** - * @returns a SASL value-safe-char encoded string, replacing "," and "=" - * with their escaped counterparts in a newly allocated string. - */ -static char *rd_kafka_sasl_safe_string (const char *str) { - char *safe = NULL, *d = NULL/*avoid warning*/; - int pass; - size_t len = 0; - - /* Pass #1: scan for needed length and allocate. - * Pass #2: encode string */ - for (pass = 0 ; pass < 2 ; pass++) { - const char *s; - for (s = str ; *s ; s++) { - if (pass == 0) { - len += 1 + (*s == ',' || *s == '='); - continue; - } - - if (*s == ',') { - *(d++) = '='; - *(d++) = '2'; - *(d++) = 'C'; - } else if (*s == '=') { - *(d++) = '='; - *(d++) = '3'; - *(d++) = 'D'; - } else - *(d++) = *s; - } - - if (pass == 0) - d = safe = rd_malloc(len+1); - } - - rd_assert(d == safe + (int)len); - *d = '\0'; - - return safe; -} - - -/** - * @brief Build client-final-message-without-proof - * @remark out->ptr will be allocated and must be freed. - */ -static void -rd_kafka_sasl_scram_build_client_final_message_wo_proof ( - struct rd_kafka_sasl_scram_state *state, - const char *snonce, - rd_chariov_t *out) { - const char *attr_c = "biws"; /* base64 encode of "n,," */ - - /* - * client-final-message-without-proof = - * channel-binding "," nonce ["," - * extensions] - */ - out->size = strlen("c=,r=") + strlen(attr_c) + - state->cnonce.size + strlen(snonce); - out->ptr = rd_malloc(out->size+1); - rd_snprintf(out->ptr, out->size+1, "c=%s,r=%.*s%s", - attr_c, (int)state->cnonce.size, state->cnonce.ptr, snonce); -} - - -/** - * @brief Build client-final-message - * @returns -1 on error. - */ -static int -rd_kafka_sasl_scram_build_client_final_message ( - rd_kafka_transport_t *rktrans, - const rd_chariov_t *salt, - const char *server_nonce, - const rd_chariov_t *server_first_msg, - int itcnt, rd_chariov_t *out) { - struct rd_kafka_sasl_scram_state *state = rktrans->rktrans_sasl.state; - const rd_kafka_conf_t *conf = &rktrans->rktrans_rkb->rkb_rk->rk_conf; - rd_chariov_t SaslPassword = - { .ptr = conf->sasl.password, - .size = strlen(conf->sasl.password) }; - rd_chariov_t SaltedPassword = - { .ptr = rd_alloca(EVP_MAX_MD_SIZE) }; - rd_chariov_t ClientKey = - { .ptr = rd_alloca(EVP_MAX_MD_SIZE) }; - rd_chariov_t ServerKey = - { .ptr = rd_alloca(EVP_MAX_MD_SIZE) }; - rd_chariov_t StoredKey = - { .ptr = rd_alloca(EVP_MAX_MD_SIZE) }; - rd_chariov_t AuthMessage = RD_ZERO_INIT; - rd_chariov_t ClientSignature = - { .ptr = rd_alloca(EVP_MAX_MD_SIZE) }; - rd_chariov_t ServerSignature = - { .ptr = rd_alloca(EVP_MAX_MD_SIZE) }; - const rd_chariov_t ClientKeyVerbatim = - { .ptr = "Client Key", .size = 10 }; - const rd_chariov_t ServerKeyVerbatim = - { .ptr = "Server Key", .size = 10 }; - rd_chariov_t ClientProof = - { .ptr = rd_alloca(EVP_MAX_MD_SIZE) }; - rd_chariov_t client_final_msg_wo_proof; - char *ClientProofB64; - int i; - - /* Constructing the ClientProof attribute (p): - * - * p = Base64-encoded ClientProof - * SaltedPassword := Hi(Normalize(password), salt, i) - * ClientKey := HMAC(SaltedPassword, "Client Key") - * StoredKey := H(ClientKey) - * AuthMessage := client-first-message-bare + "," + - * server-first-message + "," + - * client-final-message-without-proof - * ClientSignature := HMAC(StoredKey, AuthMessage) - * ClientProof := ClientKey XOR ClientSignature - * ServerKey := HMAC(SaltedPassword, "Server Key") - * ServerSignature := HMAC(ServerKey, AuthMessage) - */ - - /* SaltedPassword := Hi(Normalize(password), salt, i) */ - if (rd_kafka_sasl_scram_Hi( - rktrans, &SaslPassword, salt, - itcnt, &SaltedPassword) == -1) - return -1; - - /* ClientKey := HMAC(SaltedPassword, "Client Key") */ - if (rd_kafka_sasl_scram_HMAC( - rktrans, &SaltedPassword, &ClientKeyVerbatim, - &ClientKey) == -1) - return -1; - - /* StoredKey := H(ClientKey) */ - if (rd_kafka_sasl_scram_H(rktrans, &ClientKey, &StoredKey) == -1) - return -1; - - /* client-final-message-without-proof */ - rd_kafka_sasl_scram_build_client_final_message_wo_proof( - state, server_nonce, &client_final_msg_wo_proof); - - /* AuthMessage := client-first-message-bare + "," + - * server-first-message + "," + - * client-final-message-without-proof */ - AuthMessage.size = - state->first_msg_bare.size + 1 + - server_first_msg->size + 1 + - client_final_msg_wo_proof.size; - AuthMessage.ptr = rd_alloca(AuthMessage.size+1); - rd_snprintf(AuthMessage.ptr, AuthMessage.size+1, - "%.*s,%.*s,%.*s", - (int)state->first_msg_bare.size, state->first_msg_bare.ptr, - (int)server_first_msg->size, server_first_msg->ptr, - (int)client_final_msg_wo_proof.size, - client_final_msg_wo_proof.ptr); - - /* - * Calculate ServerSignature for later verification when - * server-final-message is received. - */ - - /* ServerKey := HMAC(SaltedPassword, "Server Key") */ - if (rd_kafka_sasl_scram_HMAC( - rktrans, &SaltedPassword, &ServerKeyVerbatim, - &ServerKey) == -1) { - rd_free(client_final_msg_wo_proof.ptr); - return -1; - } - - /* ServerSignature := HMAC(ServerKey, AuthMessage) */ - if (rd_kafka_sasl_scram_HMAC(rktrans, &ServerKey, - &AuthMessage, &ServerSignature) == -1) { - rd_free(client_final_msg_wo_proof.ptr); - return -1; - } - - /* Store the Base64 encoded ServerSignature for quick comparison */ - state->ServerSignatureB64 = rd_base64_encode(&ServerSignature); - - - /* - * Continue with client-final-message - */ - - /* ClientSignature := HMAC(StoredKey, AuthMessage) */ - if (rd_kafka_sasl_scram_HMAC(rktrans, &StoredKey, - &AuthMessage, &ClientSignature) == -1) { - rd_free(client_final_msg_wo_proof.ptr); - return -1; - } - - /* ClientProof := ClientKey XOR ClientSignature */ - assert(ClientKey.size == ClientSignature.size); - for (i = 0 ; i < (int)ClientKey.size ; i++) - ClientProof.ptr[i] = ClientKey.ptr[i] ^ ClientSignature.ptr[i]; - ClientProof.size = ClientKey.size; - - - /* Base64 encoded ClientProof */ - ClientProofB64 = rd_base64_encode(&ClientProof); - - /* Construct client-final-message */ - out->size = client_final_msg_wo_proof.size + - strlen(",p=") + strlen(ClientProofB64); - out->ptr = rd_malloc(out->size + 1); - - rd_snprintf(out->ptr, out->size+1, - "%.*s,p=%s", - (int)client_final_msg_wo_proof.size, - client_final_msg_wo_proof.ptr, - ClientProofB64); - rd_free(ClientProofB64); - rd_free(client_final_msg_wo_proof.ptr); - - return 0; -} - - -/** - * @brief Handle first message from server - * - * Parse server response which looks something like: - * "r=fyko+d2lbbFgONR....,s=QSXCR+Q6sek8bf92,i=4096" - * - * @returns -1 on error. - */ -static int -rd_kafka_sasl_scram_handle_server_first_message (rd_kafka_transport_t *rktrans, - const rd_chariov_t *in, - rd_chariov_t *out, - char *errstr, - size_t errstr_size) { - struct rd_kafka_sasl_scram_state *state = rktrans->rktrans_sasl.state; - char *server_nonce; - rd_chariov_t salt_b64, salt; - char *itcntstr; - const char *endptr; - int itcnt; - char *attr_m; - - /* Mandatory future extension check */ - if ((attr_m = rd_kafka_sasl_scram_get_attr( - in, 'm', NULL, NULL, 0))) { - rd_snprintf(errstr, errstr_size, - "Unsupported mandatory SCRAM extension"); - rd_free(attr_m); - return -1; - } - - /* Server nonce */ - if (!(server_nonce = rd_kafka_sasl_scram_get_attr( - in, 'r', - "Server nonce in server-first-message", - errstr, errstr_size))) - return -1; - - if (strlen(server_nonce) <= state->cnonce.size || - strncmp(state->cnonce.ptr, server_nonce, state->cnonce.size)) { - rd_snprintf(errstr, errstr_size, - "Server/client nonce mismatch in " - "server-first-message"); - rd_free(server_nonce); - return -1; - } - - /* Salt (Base64) */ - if (!(salt_b64.ptr = rd_kafka_sasl_scram_get_attr( - in, 's', - "Salt in server-first-message", - errstr, errstr_size))) { - rd_free(server_nonce); - return -1; - } - salt_b64.size = strlen(salt_b64.ptr); - - /* Convert Salt to binary */ - if (rd_base64_decode(&salt_b64, &salt) == -1) { - rd_snprintf(errstr, errstr_size, - "Invalid Base64 Salt in server-first-message"); - rd_free(server_nonce); - rd_free(salt_b64.ptr); - } - rd_free(salt_b64.ptr); - - /* Iteration count (as string) */ - if (!(itcntstr = rd_kafka_sasl_scram_get_attr( - in, 'i', - "Iteration count in server-first-message", - errstr, errstr_size))) { - rd_free(server_nonce); - rd_free(salt.ptr); - return -1; - } - - /* Iteration count (as int) */ - errno = 0; - itcnt = (int)strtoul(itcntstr, (char **)&endptr, 10); - if (itcntstr == endptr || *endptr != '\0' || errno != 0 || - itcnt > 1000000) { - rd_snprintf(errstr, errstr_size, - "Invalid value (not integer or too large) " - "for Iteration count in server-first-message"); - rd_free(server_nonce); - rd_free(salt.ptr); - rd_free(itcntstr); - return -1; - } - rd_free(itcntstr); - - /* Build client-final-message */ - if (rd_kafka_sasl_scram_build_client_final_message( - rktrans, &salt, server_nonce, in, itcnt, out) == -1) { - rd_snprintf(errstr, errstr_size, - "Failed to build SCRAM client-final-message"); - rd_free(salt.ptr); - rd_free(server_nonce); - return -1; - } - - rd_free(server_nonce); - rd_free(salt.ptr); - - return 0; -} - -/** - * @brief Handle server-final-message - * - * This is the end of authentication and the SCRAM state - * will be freed at the end of this function regardless of - * authentication outcome. - * - * @returns -1 on failure - */ -static int -rd_kafka_sasl_scram_handle_server_final_message ( - rd_kafka_transport_t *rktrans, - const rd_chariov_t *in, - char *errstr, size_t errstr_size) { - struct rd_kafka_sasl_scram_state *state = rktrans->rktrans_sasl.state; - char *attr_v, *attr_e; - - if ((attr_e = rd_kafka_sasl_scram_get_attr( - in, 'e', "server-error in server-final-message", - errstr, errstr_size))) { - /* Authentication failed */ - - rd_snprintf(errstr, errstr_size, - "SASL SCRAM authentication failed: " - "broker responded with %s", - attr_e); - rd_free(attr_e); - return -1; - - } else if ((attr_v = rd_kafka_sasl_scram_get_attr( - in, 'v', "verifier in server-final-message", - errstr, errstr_size))) { - const rd_kafka_conf_t *conf; - - /* Authentication succesful on server, - * but we need to verify the ServerSignature too. */ - rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY | RD_KAFKA_DBG_BROKER, - "SCRAMAUTH", - "SASL SCRAM authentication succesful on server: " - "verifying ServerSignature"); - - if (strcmp(attr_v, state->ServerSignatureB64)) { - rd_snprintf(errstr, errstr_size, - "SASL SCRAM authentication failed: " - "ServerSignature mismatch " - "(server's %s != ours %s)", - attr_v, state->ServerSignatureB64); - rd_free(attr_v); - return -1; - } - rd_free(attr_v); - - conf = &rktrans->rktrans_rkb->rkb_rk->rk_conf; - - rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY | RD_KAFKA_DBG_BROKER, - "SCRAMAUTH", - "Authenticated as %s using %s", - conf->sasl.username, - conf->sasl.mechanisms); - - rd_kafka_sasl_auth_done(rktrans); - return 0; - - } else { - rd_snprintf(errstr, errstr_size, - "SASL SCRAM authentication failed: " - "no verifier or server-error returned from broker"); - return -1; - } -} - - - -/** - * @brief Build client-first-message - */ -static void -rd_kafka_sasl_scram_build_client_first_message ( - rd_kafka_transport_t *rktrans, - rd_chariov_t *out) { - char *sasl_username; - struct rd_kafka_sasl_scram_state *state = rktrans->rktrans_sasl.state; - const rd_kafka_conf_t *conf = &rktrans->rktrans_rkb->rkb_rk->rk_conf; - - rd_kafka_sasl_scram_generate_nonce(&state->cnonce); - - sasl_username = rd_kafka_sasl_safe_string(conf->sasl.username); - - out->size = strlen("n,,n=,r=") + strlen(sasl_username) + - state->cnonce.size; - out->ptr = rd_malloc(out->size+1); - - rd_snprintf(out->ptr, out->size+1, - "n,,n=%s,r=%.*s", - sasl_username, - (int)state->cnonce.size, state->cnonce.ptr); - rd_free(sasl_username); - - /* Save client-first-message-bare (skip gs2-header) */ - state->first_msg_bare.size = out->size-3; - state->first_msg_bare.ptr = rd_memdup(out->ptr+3, - state->first_msg_bare.size); -} - - - -/** - * @brief SASL SCRAM client state machine - * @returns -1 on failure (errstr set), else 0. - */ -static int rd_kafka_sasl_scram_fsm (rd_kafka_transport_t *rktrans, - const rd_chariov_t *in, - char *errstr, size_t errstr_size) { - static const char *state_names[] = { - "client-first-message", - "server-first-message", - "client-final-message", - }; - struct rd_kafka_sasl_scram_state *state = rktrans->rktrans_sasl.state; - rd_chariov_t out = RD_ZERO_INIT; - int r = -1; - rd_ts_t ts_start = rd_clock(); - int prev_state = state->state; - - rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY, "SASLSCRAM", - "SASL SCRAM client in state %s", - state_names[state->state]); - - switch (state->state) - { - case RD_KAFKA_SASL_SCRAM_STATE_CLIENT_FIRST_MESSAGE: - rd_dassert(!in); /* Not expecting any server-input */ - - rd_kafka_sasl_scram_build_client_first_message(rktrans, &out); - state->state = RD_KAFKA_SASL_SCRAM_STATE_SERVER_FIRST_MESSAGE; - break; - - - case RD_KAFKA_SASL_SCRAM_STATE_SERVER_FIRST_MESSAGE: - rd_dassert(in); /* Requires server-input */ - - if (rd_kafka_sasl_scram_handle_server_first_message( - rktrans, in, &out, errstr, errstr_size) == -1) - return -1; - - state->state = RD_KAFKA_SASL_SCRAM_STATE_CLIENT_FINAL_MESSAGE; - break; - - case RD_KAFKA_SASL_SCRAM_STATE_CLIENT_FINAL_MESSAGE: - rd_dassert(in); /* Requires server-input */ - - r = rd_kafka_sasl_scram_handle_server_final_message( - rktrans, in, errstr, errstr_size); - break; - } - - if (out.ptr) { - r = rd_kafka_sasl_send(rktrans, out.ptr, (int)out.size, - errstr, errstr_size); - rd_free(out.ptr); - } - - ts_start = (rd_clock() - ts_start) / 1000; - if (ts_start >= 100) - rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY, "SCRAM", - "SASL SCRAM state %s handled in %"PRId64"ms", - state_names[prev_state], ts_start); - - - return r; -} - - -/** - * @brief Handle received frame from broker. - */ -static int rd_kafka_sasl_scram_recv (rd_kafka_transport_t *rktrans, - const void *buf, size_t size, - char *errstr, size_t errstr_size) { - const rd_chariov_t in = { .ptr = (char *)buf, .size = size }; - return rd_kafka_sasl_scram_fsm(rktrans, &in, errstr, errstr_size); -} - - -/** - * @brief Initialize and start SASL SCRAM (builtin) authentication. - * - * Returns 0 on successful init and -1 on error. - * - * @locality broker thread - */ -static int rd_kafka_sasl_scram_client_new (rd_kafka_transport_t *rktrans, - const char *hostname, - char *errstr, size_t errstr_size) { - struct rd_kafka_sasl_scram_state *state; - - state = rd_calloc(1, sizeof(*state)); - state->state = RD_KAFKA_SASL_SCRAM_STATE_CLIENT_FIRST_MESSAGE; - rktrans->rktrans_sasl.state = state; - - /* Kick off the FSM */ - return rd_kafka_sasl_scram_fsm(rktrans, NULL, errstr, errstr_size); -} - - - -/** - * @brief Validate SCRAM config and look up the hash function - */ -static int rd_kafka_sasl_scram_conf_validate (rd_kafka_t *rk, - char *errstr, - size_t errstr_size) { - const char *mech = rk->rk_conf.sasl.mechanisms; - - if (!rk->rk_conf.sasl.username || !rk->rk_conf.sasl.password) { - rd_snprintf(errstr, errstr_size, - "sasl.username and sasl.password must be set"); - return -1; - } - - if (!strcmp(mech, "SCRAM-SHA-1")) { - rk->rk_conf.sasl.scram_evp = EVP_sha1(); - rk->rk_conf.sasl.scram_H = SHA1; - rk->rk_conf.sasl.scram_H_size = SHA_DIGEST_LENGTH; - } else if (!strcmp(mech, "SCRAM-SHA-256")) { - rk->rk_conf.sasl.scram_evp = EVP_sha256(); - rk->rk_conf.sasl.scram_H = SHA256; - rk->rk_conf.sasl.scram_H_size = SHA256_DIGEST_LENGTH; - } else if (!strcmp(mech, "SCRAM-SHA-512")) { - rk->rk_conf.sasl.scram_evp = EVP_sha512(); - rk->rk_conf.sasl.scram_H = SHA512; - rk->rk_conf.sasl.scram_H_size = SHA512_DIGEST_LENGTH; - } else { - rd_snprintf(errstr, errstr_size, - "Unsupported hash function: %s " - "(try SCRAM-SHA-512)", - mech); - return -1; - } - - return 0; -} - - - - -const struct rd_kafka_sasl_provider rd_kafka_sasl_scram_provider = { - .name = "SCRAM (builtin)", - .client_new = rd_kafka_sasl_scram_client_new, - .recv = rd_kafka_sasl_scram_recv, - .close = rd_kafka_sasl_scram_close, - .conf_validate = rd_kafka_sasl_scram_conf_validate, -};
