Librdkafka: setting up ssl on librdkafka

[MustakMU]

Description

Hi... I'm new to kafka so please guide me on how to import ssl certificate on librdkafka c client..
These are things i've done..
I'm able to run the console producer and consumer by using client truststore and keystore files as explained in this article https://www.confluent.io/blog/apache-kafka-security-authorization-authentication-encryption/
but i'm not getting how to import these truststore and keystore file in my c client.

Checklist

Please provide the following information:

• [ ] librdkafka version (release number or git tag): 0.9.2
• [ ] Apache Kafka version: kafka_2.10-0.10.0.0
• [ ] librdkafka client configuration:
• [ ] Operating system: ubuntu 16.04
• [ ] Using the legacy Consumer:
• [ ] Using the high-level KafkaConsumer
• [ ] Provide logs (with debug=.. as necessary) from librdkafka
• [ ] Provide broker log excerpts
• [ ] Critical issue

[edenhill]

Did you see this guide?
https://github.com/edenhill/librdkafka/wiki/Using-SSL-with-librdkafka

[MustakMU]

thank you very much for reply...
ya i saw the guide... i configured broker but how do i configure librdkafka client? in which i should make the changes?

[MustakMU]

i tried to run ./rdkafka_example -X dump to see the properties here is the output ..

Global config

builtin.features = gzip,snappy,ssl,regex,lz4
client.id = rdkafka
message.max.bytes = 1000000
message.copy.max.bytes = 65535
receive.message.max.bytes = 100000000
max.in.flight.requests.per.connection = 1000000
metadata.request.timeout.ms = 60000
topic.metadata.refresh.interval.ms = 300000
metadata.max.age.ms = -1
topic.metadata.refresh.fast.interval.ms = 250
topic.metadata.refresh.fast.cnt = 10
topic.metadata.refresh.sparse = true
socket.timeout.ms = 60000
socket.blocking.max.ms = 1000
socket.send.buffer.bytes = 0
socket.receive.buffer.bytes = 0
socket.keepalive.enable = false
socket.nagle.disable = false
socket.max.fails = 3
broker.address.ttl = 1000
broker.address.family = any
reconnect.backoff.jitter.ms = 500
statistics.interval.ms = 0
enabled_events = 0
log_cb = 0x405c30
log_level = 6
log.queue = false
log.thread.name = true
log.connection.close = true
socket_cb = 0x41bd10
open_cb = 0x42af60
internal.termination.signal = 29
api.version.request = false
api.version.fallback.ms = 1200000
broker.version.fallback = 0.9.0
security.protocol = plaintext
partition.assignment.strategy = range,roundrobin
session.timeout.ms = 30000
heartbeat.interval.ms = 1000
group.protocol.type = consumer
coordinator.query.interval.ms = 600000
enable.auto.commit = true
auto.commit.interval.ms = 5000
enable.auto.offset.store = true
queued.min.messages = 100000
queued.max.messages.kbytes = 1000000
fetch.wait.max.ms = 100
fetch.message.max.bytes = 1048576
fetch.min.bytes = 1
fetch.error.backoff.ms = 500
offset.store.method = broker
enable.partition.eof = true
queue.buffering.max.messages = 100000
queue.buffering.max.kbytes = 4000000
queue.buffering.max.ms = 1000
message.send.max.retries = 2
retry.backoff.ms = 100
compression.codec = none
batch.num.messages = 10000
delivery.report.only.error = false

Topic config

request.required.acks = 1
request.timeout.ms = 5000
message.timeout.ms = 300000
produce.offset.report = false
compression.codec = inherit
auto.commit.enable = true
auto.commit.interval.ms = 60000
auto.offset.reset = largest
offset.store.path = .
offset.store.sync.interval.ms = -1
offset.store.method = broker
consume.callback.max.messages = 0

i don't see any ssl configuration properties...

[edenhill]

dump will only show set properties and properties with default settings, but none of the ssl properties have default settings so they wont show up here.
See the full property list here:
https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md

[MustakMU]

so where do i need to update these settings?
metadata.broker.list=at_least_one_of_the_brokers
security.protocol=ssl

CA certificate file for verifying the broker's certificate.

ssl.ca.location=ca-cert

Client's certificate

ssl.certificate.location=client_?????_client.pem

Client's key

ssl.key.location=client_?????_client.key

Key password, if any.

ssl.key.password=abcdefgh

[edenhill]

If you have your own program then those properties are set with:

• C: rd_kafka_conf_set()
• C++: Conf->set()

If you are using kafkacat or rdkafka_example you pass the properties as -X key=value

[MustakMU]

Thank you.. supplied all above properties to rdkafka_example and it's working...

[edenhill]

Great!

[lday00]

I have the kafka server, producer and consumer setup (3 separate Linux machines). I am using the kafka-cpp/examples/rdkafka_example.cpp to run both producer and consumer. The plaintext works fine.
Then I recompile librdkafka having WITH_SSL set to 1 and did the same client configuration like above for ssl. I don't get any error, but sending & flush in SSL doesn't work and timeout. How do I troubleshoot? I am following the instruction: https://github.com/edenhill/librdkafka/wiki/Using-SSL-with-librdkafka.

My client configuration looks like this:

    cr = conf->set("security.protocol", "SSL", errstr);
    if (cr != RD_KAFKA_CONF_OK)
    {
        cerr << "ERROR: " << errstr;
        exit(1);
    }
    cr = conf->set("ssl.ca.location", "ca-cert", errstr);
    if (cr != RD_KAFKA_CONF_OK)
    {
        cerr << "ERROR: " << errstr;
        exit(1);
    }
    cr = conf->set("ssl.certificate.location",
            "/data/adis/kafka/client_????????_client.pem", errstr);
    if (cr != RD_KAFKA_CONF_OK)
    {
        cerr << "ERROR: " << errstr;
        exit(1);
    }
    cr = conf->set("ssl.key.location",
            "/data/adis/kafka/client_????????_client.key", errstr);
    if (cr != RD_KAFKA_CONF_OK)
    {
        cerr << "ERROR: " << errstr;
        exit(1);
    }
    cr = conf->set("ssl.key.password", "abcdefgh", errstr);
    if (cr != RD_KAFKA_CONF_OK)
    {
        cerr << "ERROR: " << errstr;
        exit(1);
    }

% Produced message (2622 bytes)
% Produced message (2622 bytes)
% Producer flush failed: Local: Timed out

[lday00]

One more question is that the API ProducerImpl::produce():

RdKafka::ErrorCode RdKafka::ProducerImpl::produce (RdKafka::Topic *topic,
int32_t partition,
int msgflags,
void *payload, size_t len,
const std::string *key,
void *msg_opaque)

What should I put in parameter key for SSL? Or it should still be NULL?

[edenhill]

Enable debug=security,broker to troubleshoot further.
Also make sure you are connecting to the broker's SSL listener port and not the PLAINTEXT

[lday00]

Here is the debug log looks like:

LOG-7-SSL: [thrd:app]: Loading CA certificate(s) from file /data/adis/kafka/ca-cert
LOG-7-SSL: [thrd:app]: Loading certificate from file /data/adis/kafka/client_????????_client.pem
LOG-7-SSL: [thrd:app]: Loading private key file from /data/adis/kafka/client_????????_client.key
LOG-7-SSLPASSWD: [thrd:app]: Private key file "/data/adis/kafka/client_????????_client.key" requires password
LOG-7-BRKMAIN: [thrd::0/internal]: :0/internal: Enter main broker thread
LOG-7-STATE: [thrd::0/internal]: :0/internal: Broker changed state INIT -> UP
LOG-7-BROKER: [thrd:app]: ssl://fqdn.name:9093/bootstrap: Added new broker with NodeId -1
% Created producer rdkafka#producer-1
LOG-7-BRKMAIN: [thrd:ssl://fqdn.name:9093/bootstrap]: ssl://fqdn.name:9093/bootstrap: Enter main broker thread
LOG-7-CONNECT: [thrd:ssl://fqdn.name:9093/bootstrap]: ssl://fqdn.name:9093/bootstrap: broker in state INIT connecting
LOG-7-CONNECT: [thrd:ssl://fqdn.name:9093/bootstrap]: ssl://fqdn.name:9093/bootstrap: Connecting to ipv4#1.2.3.4:9093 (ssl) with socket 8
LOG-7-STATE: [thrd:ssl://fqdn.name:9093/bootstrap]: ssl://fqdn.name:9093/bootstrap: Broker changed state INIT -> CONNECT
LOG-7-CONNECT: [thrd:ssl://fqdn.name:9093/bootstrap]: ssl://fqdn.name:9093/bootstrap: Connected to ipv4#1.2.3.4:9093

[lday00]

Still having the same issue:
% Produced message (2622 bytes)
% Producer flush failed: Local: Timed out

I am looking at https://github.com/edenhill/librdkafka/wiki/Using-SSL-with-librdkafka. Do all gen-ssl-certs.sh commands (certificate, broker keystore, client keys) run from the kafka server box or they need to be run from different boxes?

[edenhill]

Doesn't matter where you generate the certs, but they need to be copied to the hosts that are to use them.
Also read this: https://docs.confluent.io/current/security.html#security

[lday00]

I troubleshoot and found out SSL_connect() failed (returned -1), but code does not check for it. I added the check, the returned value is -1 and SSL_get_error() returned 2 (SSL_ERROR_WANT_READ).

rdkafka_transport.c: 671
r = SSL_connect(rktrans->rktrans_ssl);
if (r == 1) {
/* Connected, highly unlikely since this is a
* non-blocking operation. */
rd_kafka_transport_connect_done(rktrans, NULL);
return 0;
}

else
{
r = SSL_get_error(rktrans->rktrans_ssl, r));
printf("SSL_get_error(): %d\n", r)
return -1;
}

[edenhill]

And you are absolutely sure you are connecting to the SSL listener of the broker?

[edenhill]

From https://linux.die.net/man/3/ssl_connect, re return values:

2.
The TLS/SSL handshake was not successful but was shut down controlled and by the specifications of the TLS/SSL protocol. Call SSL_get_error() with the return value ret to find out the reason.

But this does not seem to be documented for OpenSSL 1.0.x.
What OpenSSL version are you using?

[devloper090909]

Hi,

I am trying to do SSL configuration for kafka but in the beginning of configuration setting I am facing one error :
_Failed to set Kafka security.protocol = ssl returned -1 error Invalid value for configuration property "security.protocol"_

 cRes = conf->set("security.protocol", "ssl", errStr);
    switch (cRes) {
    case RdKafka::Conf::CONF_OK:
        break;
    case RdKafka::Conf::CONF_INVALID:
    case RdKafka::Conf::CONF_UNKNOWN:
    default:
        if (resmon::dbgTrace) {
            RESMON_LOG(LOG_ERR, "Failed to set Kafka security.protocol = ssl "
                       "returned %d error %s\n",
                       cRes, errStr.c_str());
        }

    }

[rolandyoung]

Did you check that librdkafka has been built with SSL support? You should also check all the other prerequisites listed here: https://github.com/edenhill/librdkafka/wiki/Using-SSL-with-librdkafka

[devloper090909]

this is config.h

// Automatically generated by ./configure
#pragma once
#define ARCH "x86_64"
#define CPU "generic"
#define WITHOUT_OPTIMIZATION 0
#define ENABLE_DEVEL 0
#define ENABLE_VALGRIND 0
#define ENABLE_REFCNT_DEBUG 0
#define ENABLE_SHAREDPTR_DEBUG 0
#define ENABLE_LZ4_EXT 0
**#define ENABLE_SSL 1**
#define ENABLE_SASL 0
#define WITH_STATIC_LINKING 1
#define ENABLE_LZ4_EXT 0
#define ENABLE_SSL 1
#define ENABLE_SASL 0
#define MKL_APP_NAME "librdkafka"
#define MKL_APP_DESC_ONELINE "The Apache Kafka C/C++ library"
// gcc
#define WITH_GCC 1
// gxx
#define WITH_GXX 1
// pkgconfig
#define WITH_PKGCONFIG 1
// install
#define WITH_INSTALL 1
// PIC
#define HAVE_PIC 1
// gnulib
#define WITH_GNULD 1
// __atomic_32
#define HAVE_ATOMICS_32 1
// __atomic_32
#define HAVE_ATOMICS_32_ATOMIC 1
// atomic_32
#define ATOMIC_OP32(OP1,OP2,PTR,VAL) __atomic_ ## OP1 ## _ ## OP2(PTR, VAL, __ATOMIC_SEQ_CST)
// __atomic_64
#define HAVE_ATOMICS_64 1
// __atomic_64
#define HAVE_ATOMICS_64_ATOMIC 1
// atomic_64
#define ATOMIC_OP64(OP1,OP2,PTR,VAL) __atomic_ ## OP1 ## _ ## OP2(PTR, VAL, __ATOMIC_SEQ_CST)
// atomic_64
#define ATOMIC_OP(OP1,OP2,PTR,VAL) __atomic_ ## OP1 ## _ ## OP2(PTR, VAL, __ATOMIC_SEQ_CST)
// parseversion
#define RDKAFKA_VERSION_STR "0.9.5"
// parseversion
#define MKL_APP_VERSION "0.9.5"
// WITH_SNAPPY
#define WITH_SNAPPY 1
// WITH_SOCKEM
#define WITH_SOCKEM 1
// regex
#define HAVE_REGEX 1
// strndup
#define HAVE_STRNDUP 1
// python
#define HAVE_PYTHON 1

[rolandyoung]

If you look at rdkafka_conf.c, you will see that "ssl" is only recognised as a protocol option if WITH_SSL is defined at build time. Since ENABLE_SSL is defined, you have asked for SSL support at configure time, but it has not been included in the configuration.

I would guess that one of the prerequisite libraries is missing from your build environment, probably libssl-dev. You should have seen a warning message when you ran ./configure.

[devloper090909]

./configure

checking for libssl (by pkg-config)... failed ----> How can I get it

using cache file config.cache
checking for OS or distribution... ok (Ubuntu)
checking for C compiler from CC env... failed
checking for gcc (by command)... ok (cached)
checking for C++ compiler from CXX env... failed
checking for C++ compiler (g++)... ok (cached)
checking executable ld... ok (cached)
checking executable nm... ok (cached)
checking executable objdump... ok (cached)
checking executable strip... ok (cached)
checking for pkgconfig (by command)... ok (cached)
checking for install (by command)... ok (cached)
checking for PIC (by compile)... ok (cached)
checking for GNU-compatible linker options... ok (cached)
checking for GNU linker-script ld flag... ok (cached)
checking for __atomic_32 (by compile)... ok (cached)
checking for __atomic_64 (by compile)... ok (cached)
checking for socket (by compile)... ok (cached)
parsing version '0x000905ff'... ok (0.9.5)
checking for libpthread (by pkg-config)... failed
checking for libpthread (by compile)... ok (cached)
checking for libcrypto (by pkg-config)... failed
checking for libcrypto (by compile)... failed (disable)
checking for liblz4 (by pkg-config)... failed
checking for liblz4 (by compile)... failed (disable)
checking for libssl (by pkg-config)... failed
checking for libssl (by compile)... failed (disable)
checking for libsasl2 (by pkg-config)... failed
checking for libsasl2 (by compile)... failed (disable)
checking for libsasl (by pkg-config)... failed
checking for libsasl (by compile)... failed (disable)
checking for regex (by compile)... ok (cached)
checking for librt (by pkg-config)... failed
checking for librt (by compile)... ok (cached)
checking for strndup (by compile)... ok (cached)
checking for nm (by env NM)... ok (cached)
checking for python (by command)... ok (cached)
Generated Makefile.config
Generated config.h

Configuration summary:
prefix /usr/local
ARCH x86_64
CPU generic
GEN_PKG_CONFIG y
ENABLE_DEVEL n
ENABLE_VALGRIND n
ENABLE_REFCNT_DEBUG n
ENABLE_SHAREDPTR_DEBUG n
ENABLE_LZ4_EXT y
ENABLE_SSL y
ENABLE_SASL y
MKL_APP_NAME librdkafka
MKL_APP_DESC_ONELINE The Apache Kafka C/C++ library
MKL_DISTRO Ubuntu
CC gcc
CXX g++
LD ld
NM nm
OBJDUMP objdump
STRIP strip
CPPFLAGS -g -O2 -fPIC -Wall -Wsign-compare -Wfloat-equal -Wpointer-arith -Wcast-align
PKG_CONFIG pkg-config
INSTALL install
LIB_LDFLAGS -shared -Wl,-soname,$(LIBFILENAME)
LDFLAG_LINKERSCRIPT -Wl,--version-script=
RDKAFKA_VERSION_STR 0.9.5
MKL_APP_VERSION 0.9.5
LIBS -lpthread -lrt
CXXFLAGS -Wno-non-virtual-dtor
SYMDUMPER $(NM) -D
exec_prefix /usr/local
bindir /usr/local/bin
sbindir /usr/local/sbin
libexecdir /usr/local/libexec
datadir /usr/local/share
sysconfdir /usr/local/etc
sharedstatedir /usr/local/com
localstatedir /usr/local/var
libdir /usr/local/lib
includedir /usr/local/include
infodir /usr/local/info
mandir /usr/local/man
Generated config.cache

Now type 'make' to build

[rolandyoung]

Try the query "ubuntu install libssl-dev" in your preferred search engine, replacing 'ubuntu' with the name of your operating system.