Python kafka confluent SSL config using keytool pem file

Question

I have this keytool bash file that loads my certs:

regions=( us-east us-west )
CACERT_FILE="${JAVA_HOME}/lib/security/cacerts"
mkdir /keys

echo "Loading Certs"

curl -s <url> > /keys/root_ca.pem
keytool -importcert -keystore ${CACERT_FILE} -alias root -storepass <pass> -file /keys/root_ca.pem -trustcacerts -noprompt

for i in "${regions[@]}"
do
  echo "Importing intermediate CA from $i"
  curl -s <region url> > /keys/${i}.pem
  keytool -importcert -keystore ${CACERT_FILE} -alias ${i} -storepass <pass> -file /keys/${i}.pem -trustcacerts -noprompt
done

echo "Done loading certs"

Using these commands what is the right config for confluent kafka?

Also I dont know if it would help but I think the PEM uses tls 1.2

Answer 1

right config -

    ssl_cafile=/keys/root_ca.pem
    ssl_certfile=/keys/${i}.pem
    ssl_keyfile=/keys/user.key.pem
    ssl_password=<key password>

You would need to replace ${i} with the appropriate region name. For example, if you were connecting to the us-east region, you would use the following config:

    ssl_certfile=/keys/us-east.pem

The ssl_cafile parameter points to the root CA certificate file. The ssl_certfile parameter points to the intermediate CA certificate file for the region you are connecting to. The ssl_keyfile parameter points to the user's certificate file. The ssl_password parameter is the password for the user's certificate file.

refernce - Python Confluent-Kafka SSL Configuration

https://docs.confluent.io/platform/current/kafka/authentication_ssl.html The PEM file uses TLS 1.2.

check the above process and i think it should be working

Answer 2

The script starts by creating the /keys/ directory, then import certs and set aliases. Supposing that the server is configured with SSL, for us-east, here is an example of a python consumer:

from confluent_kafka import Consumer

# Kafka broker URL for us-east region
bootstrap_servers = 'your_us_east_kafka_broker_url:9092'

# Consumer group 
group_id = 'your_consumer_group_id'

 
    # SSL configurations
ssl_ca_location = '/keys/root_ca.pem'
ssl_cert_location = '/keys/us-east.pem'
ssl_key_location = '/keys/us-east.key.pem'  # Replace with the correct key file 
ssl_password = 'your_keystore_password'  # Replace with the correct pwd

    
consumer_config = {
    'bootstrap.servers': bootstrap_servers,
    'group.id': group_id,
    'security.protocol': 'ssl',
    'ssl.ca.location': ssl_ca_location,
    'ssl.certificate.location': ssl_cert_location,
    'ssl.key.location': ssl_key_location,
    'ssl.key.password': ssl_password,
    'auto.offset.reset': 'earliest',  # Adjust as needed
    'enable.auto.commit': False  # Disable auto-commit to have control over committing offsets
}

# Create Kafka consumer
consumer = Consumer(consumer_config)

# Subscribe to topics
topics = ['your_topic']
consumer.subscribe(topics)

# Consume messages
...