SASL_SSL connection without using SSL Client Certificate Auth

[unverified-contact]

I am trying to be able to establish a connection to a broker while using SASL_SSL without using SSL Client Certificate Auth.

Firstly, I'll explain that I'm new to Kafka so I have tried to be as absolutely detailed as possible here so there should be little confusion by the end of this.

This would be connection to my broker which would work without supplying this information in the configuration: (as taken from the example at https://github.com/weiboad/kafka-php/blob/master/example/Producer.php)

$config->setSaslKeytab('/etc/security/keytabs/kafkaclient.keytab');
$config->setSaslPrincipal('kafka/node1@NMREDKAFKA.COM');
// if use ssl connect
$config->setSslLocalCert('/home/vagrant/code/kafka-php/ca-cert');
$config->setSslLocalPk('/home/vagrant/code/kafka-php/ca-key');
$config->setSslPassphrase('123456');
$config->setSslPeerName('nmred');

I don't want to be using this CA functionality nor am I looking to provide SSL credentials. Here's what I'm trying... Not sure if I'm doing it quite right but here was my attempt:

<?php

require __DIR__ . '/vendor/autoload.php';

use Kafka\Config;
use Kafka\Producer;
use Kafka\ProducerConfig;

function produce($topic, $value)
{
    $config = ProducerConfig::getInstance();

    $config->setMetadataRefreshIntervalMs(1000);
    $config->setMetadataBrokerList(getenv('BROKERS'));
    $config->setBrokerVersion('2.2.0');
    $config->setRequiredAck(1);
    $config->setIsAsyn(false);
    $config->setProduceInterval(500);

    $config->setSecurityProtocol(Config::SECURITY_PROTOCOL_SASL_SSL);
    $config->setSaslUsername(getenv('USERNAME'));
    $config->setSaslPassword(getenv('PASSWORD'));
    $config->setSaslMechanism(Config::SASL_MECHANISMS_PLAIN);

    $producer = new Producer();

    $x =  $producer->send([
        [
            'topic' => $topic,
            'value' => $value
        ],
    ]);

    var_dump($x);
}

produce('my_topic', time(). ' AAAAAA');

So when I try to run the above code, I get a bunch of errors:

PHP Warning:  stream_socket_client(): Unable to set local cert chain file `'; Check that your cafile/capath settings include details of your certificate and its issuer in /home/moth/kafkatesting/vendor/nmred/kafka-php/src/CommonSocket.php on line 217
PHP Warning:  stream_socket_client(): Failed to enable crypto in /home/moth/kafkatesting/vendor/nmred/kafka-php/src/CommonSocket.php on line 189
PHP Warning:  stream_socket_client(): unable to connect to ssl://cp69.ap-southeast-2.aws.confluent.cloud:9092 (Unknown error) in /home/moth/kafkatesting/vendor/nmred/kafka-php/src/CommonSocket.php on line 217
PHP Fatal error:  Uncaught Kafka\Exception\ConnectionException: It was not possible to establish a connection for metadata with the brokers "edited out:{{MY_BROKER_HERE}}" in /home/moth/kafkatesting/vendor/nmred/kafka-php/src/Exception/ConnectionException.php:13
Stack trace:
#0 /home/moth/kafkatesting/vendor/nmred/kafka-php/src/Producer/SyncProcess.php(146): Kafka\Exception\ConnectionException::fromBrokerList('cp69.ap-southea...')
#1 /home/moth/kafkatesting/vendor/nmred/kafka-php/src/Producer/SyncProcess.php(38): Kafka\Producer\SyncProcess->syncMeta()
#2 /home/moth/kafkatesting/vendor/nmred/kafka-php/src/Producer.php(24): Kafka\Producer\SyncProcess->__construct()
#3 /home/moth/kafkatesting/aaa.php(25): Kafka\Producer->__construct()
#4 /home/moth/kafkatesting/aaa.php(38): produce('my_topic', '1559183546AAAAA...')
#5 {main}
  thrown in /home/moth/kafkatesting/vendor/nmred/kafka-php/src/Exception/ConnectionException.php on line 13

In the code path taken here stream_socket_client() throws a Warning because it was given a $context earlier, which seems to have expected that the cert related values are set. See createStream() in src/CommonSocket.php https://github.com/weiboad/kafka-php/blob/a8f5b01d9ca24c183b121d624e3402bf8aa70488/src/CommonSocket.php#L126

$context = stream_context_create(
    [
        'ssl' => [
            'local_cert'  => $this->config->getSslLocalCert(),
            'local_pk'    => $this->config->getSslLocalPk(),
            'verify_peer' => $this->config->getSslVerifyPeer(),
            'passphrase'  => $this->config->getSslPassphrase(),
            'cafile'      => $this->config->getSslCafile(),
            'peer_name'   => $this->config->getSslPeerName(),
        ],
    ]
);

... but in my case I don't even want to be using those so I hadn't set them. So they are populated with empty or otherwise invalid defaults, causing the errors we see later.

Now, I am able to workaround this problem very simply, and get the producer working exactly as I'd expect, by changing the createSocket() function to work like this:

protected function createSocket(string $remoteSocket, $context, ?int &$errno, ?string &$errstr)
{
    return stream_socket_client(
        $remoteSocket,
        $errno,
        $errstr,
        100000,
        STREAM_CLIENT_CONNECT
        //$context
    );
}

This change will result in no errors and a producer that seems to work exactly the way I'm trying for. By not passing in the context with the above details, the connection seems to work perfectly... At least in that the message successfully gets written to the broker.

I realise it's not relevant to this particular repo however I have been able to get the producer functioning in NodeJS with the kafka-node library this way... It's pretty straightforward so I'm including it just for reference to show how the connection is made without the CA stuff and without SSL credentials being specified:

const HighLevelProducer = require('kafka-node').HighLevelProducer;
const Client = require('kafka-node').KafkaClient;

let clientOptions = {};
clientOptions.kafkaHost = KAFKA_BROKERS;
clientOptions.connectTimeout = 3000;
clientOptions.requestTimeout = 3000;
clientOptions.ssl = true;
clientOptions.sslOptions = { rejectUnauthorized: true };
clientOptions.sasl = { mechanism: 'plain', username: USERNAME, password: PASSWORD };

const client = new Client(clientOptions);
var producer = new HighLevelProducer(client);

producer.send([{ topic: 'my_topic', messages: [JSON.stringify(result)] }], function (err, data) {
    if (err) console.log(err);
});

For my broker, I'm also able to use kafkacat without specifying these additional details either like this:

kafkacat -P -b {BROKERS} -X sasl.mechanism=PLAIN -X request.timeout.ms=20000 -X retry.backoff.ms=500 -X sasl.username="{USERNAME}" -X sasl.password="{PASSWORD}" -X security.protocol=SASL_SSL -t my_topic

So my questions now are... is this is a bug? Or am I doing something wrong in my setup? Is this library even intended to work to support the connection I'd like? And if not, is that because it's considered bad practice or something like that? At bare minimum I wanted raise this as an issue because it really seems like with some minor changes this could easily support what I'm trying to do right out of the box without any changes to the lib but I don't have the familiarity to assert with confidence that I haven't done anything wrong here.

[unverified-contact]

Okay, some notes regarding my findings on this and what solution I went with.

I ended up downloading and specifying the CA Cert file from Mozilla You can find them here:

https://curl.haxx.se/ca/cacert.pem <--- this link should always be the latest one

More information on that file: https://curl.haxx.se/docs/caextract.html

I need to caution as a caveat, I'm very inexperienced with SSL. I don't know the full implications for security that this might entail with regards to someone else's application but for my purposes it works fine.

My guess (but I am not totally sure) is that the code I mentioned in my prior comment makes an assumption that a CA Cert file will be provided here because it may be considered best practice, for security purposes, to explicitly source and provide such a file.