/*
* 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
};