Running Kafka using Docker, SSL options don't appear to be used correctly-CodePudding

I have the following docker-compose.yml file in order to run a single instance of zookeeper and one broker. I thought I was having all sorts of issues with SSL handshakes, but then when I dug deeper it appears that its only paying attention to a small subset of the configuration options that I'm passing through.

version: '3.5'
networks:
  secNet:
    name: secNet

services:
  sec-zookeeper:
    image: confluentinc/cp-zookeeper:5.5.0
    container_name: sec-zookeeper
    networks:
      - secNet
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  sec-broker-one:
    image: confluentinc/cp-kafka:5.5.0
    container_name: sec-broker-one
    networks:
      - secNet
    restart: always
    ports:
      - "19092:19092"
    depends_on:
      - sec-zookeeper
    volumes:
      - ./ssl/:/etc/kafka/secrets
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: sec-zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://sec-broker-one:9092,SSL://sec-broker-one:9192
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://sec-broker-one:9092,CONNECTIONS_FROM_HOST://localhost:19092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONNECTIONS_FROM_HOST:SSL
      KAFKA_LOG_MESSAGE_TIMESTAMP_TYPE: CreateTime
      KAFKA_LOG_RETENTION_HOURS: 2232
      KAFKA_SSL_KEYSTORE_FILENAME: kafka.server.keystore.jks
      KAFKA_SSL_KEYSTORE_CREDENTIALS: server.password
      KAFKA_SSL_KEY_CREDENTIALS: server.password
      KAFKA_SSL_TRUSTSTORE_FILENAME: kafka.server.truststore.jks
      KAFKA_SSL_TRUSTSTORE_CREDENTIALS: server.password
      KAFKA_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM: ""
      KAFKA_OPTS: -Djavax.net.debug=all      
      KAFKA_SSL_CLIENT_AUTH: required

If we have a look at the results of docker-compose up output:

sec-broker-one    | ===> Launching kafka ... 
sec-broker-one    | [2022-06-10 05:14:22,327] INFO Registered kafka:type=kafka.Log4jController MBean (kafka.utils.Log4jControllerRegistration$)
sec-broker-one    | [2022-06-10 05:14:22,738] INFO KafkaConfig values: 
sec-broker-one    |     advertised.host.name = null
sec-broker-one    |     advertised.listeners = PLAINTEXT://sec-broker-one:9092,CONNECTIONS_FROM_HOST://localhost:19092
sec-broker-one    |     advertised.port = null
sec-broker-one    |     alter.config.policy.class.name = null
sec-broker-one    |     alter.log.dirs.replication.quota.window.num = 11
sec-broker-one    |     alter.log.dirs.replication.quota.window.size.seconds = 1
sec-broker-one    |     authorizer.class.name = 
sec-broker-one    |     auto.create.topics.enable = true
sec-broker-one    |     auto.leader.rebalance.enable = true
sec-broker-one    |     background.threads = 10
sec-broker-one    |     broker.id = 1
sec-broker-one    |     broker.id.generation.enable = true
sec-broker-one    |     broker.rack = null
sec-broker-one    |     client.quota.callback.class = null
sec-broker-one    |     compression.type = producer
sec-broker-one    |     connection.failed.authentication.delay.ms = 100
sec-broker-one    |     connections.max.idle.ms = 600000
sec-broker-one    |     connections.max.reauth.ms = 0
sec-broker-one    |     control.plane.listener.name = null
sec-broker-one    |     controlled.shutdown.enable = true
sec-broker-one    |     controlled.shutdown.max.retries = 3
sec-broker-one    |     controlled.shutdown.retry.backoff.ms = 5000
sec-broker-one    |     controller.socket.timeout.ms = 30000
sec-broker-one    |     create.topic.policy.class.name = null
sec-broker-one    |     default.replication.factor = 1
sec-broker-one    |     delegation.token.expiry.check.interval.ms = 3600000
sec-broker-one    |     delegation.token.expiry.time.ms = 86400000
sec-broker-one    |     delegation.token.master.key = null
sec-broker-one    |     delegation.token.max.lifetime.ms = 604800000
sec-broker-one    |     delete.records.purgatory.purge.interval.requests = 1
sec-broker-one    |     delete.topic.enable = true
sec-broker-one    |     fetch.max.bytes = 57671680
sec-broker-one    |     fetch.purgatory.purge.interval.requests = 1000
sec-broker-one    |     group.initial.rebalance.delay.ms = 3000
sec-broker-one    |     group.max.session.timeout.ms = 1800000
sec-broker-one    |     group.max.size = 2147483647
sec-broker-one    |     group.min.session.timeout.ms = 6000
sec-broker-one    |     host.name = 
sec-broker-one    |     inter.broker.listener.name = null
sec-broker-one    |     inter.broker.protocol.version = 2.5-IV0
sec-broker-one    |     kafka.metrics.polling.interval.secs = 10
sec-broker-one    |     kafka.metrics.reporters = []
sec-broker-one    |     leader.imbalance.check.interval.seconds = 300
sec-broker-one    |     leader.imbalance.per.broker.percentage = 10
sec-broker-one    |     listener.security.protocol.map = PLAINTEXT:PLAINTEXT,CONNECTIONS_FROM_HOST:SSL
sec-broker-one    |     listeners = PLAINTEXT://0.0.0.0:9092,CONNECTIONS_FROM_HOST://0.0.0.0:19092
sec-broker-one    |     log.cleaner.backoff.ms = 15000
sec-broker-one    |     log.cleaner.dedupe.buffer.size = 134217728
sec-broker-one    |     log.cleaner.delete.retention.ms = 86400000
sec-broker-one    |     log.cleaner.enable = true
sec-broker-one    |     log.cleaner.io.buffer.load.factor = 0.9
sec-broker-one    |     log.cleaner.io.buffer.size = 524288
sec-broker-one    |     log.cleaner.io.max.bytes.per.second = 1.7976931348623157E308
sec-broker-one    |     log.cleaner.max.compaction.lag.ms = 9223372036854775807
sec-broker-one    |     log.cleaner.min.cleanable.ratio = 0.5
sec-broker-one    |     log.cleaner.min.compaction.lag.ms = 0
sec-broker-one    |     log.cleaner.threads = 1
sec-broker-one    |     log.cleanup.policy = [delete]
sec-broker-one    |     log.dir = /tmp/kafka-logs
sec-broker-one    |     log.dirs = /var/lib/kafka/data
sec-broker-one    |     log.flush.interval.messages = 9223372036854775807
sec-broker-one    |     log.flush.interval.ms = null
sec-broker-one    |     log.flush.offset.checkpoint.interval.ms = 60000
sec-broker-one    |     log.flush.scheduler.interval.ms = 9223372036854775807
sec-broker-one    |     log.flush.start.offset.checkpoint.interval.ms = 60000
sec-broker-one    |     log.index.interval.bytes = 4096
sec-broker-one    |     log.index.size.max.bytes = 10485760
sec-broker-one    |     log.message.downconversion.enable = true
sec-broker-one    |     log.message.format.version = 2.5-IV0
sec-broker-one    |     log.message.timestamp.difference.max.ms = 9223372036854775807
sec-broker-one    |     log.message.timestamp.type = CreateTime
sec-broker-one    |     log.preallocate = false
sec-broker-one    |     log.retention.bytes = -1
sec-broker-one    |     log.retention.check.interval.ms = 300000
sec-broker-one    |     log.retention.hours = 2232
sec-broker-one    |     log.retention.minutes = null
sec-broker-one    |     log.retention.ms = null
sec-broker-one    |     log.roll.hours = 168
sec-broker-one    |     log.roll.jitter.hours = 0
sec-broker-one    |     log.roll.jitter.ms = null
sec-broker-one    |     log.roll.ms = null
sec-broker-one    |     log.segment.bytes = 1073741824
sec-broker-one    |     log.segment.delete.delay.ms = 60000
sec-broker-one    |     max.connections = 2147483647
sec-broker-one    |     max.connections.per.ip = 2147483647
sec-broker-one    |     max.connections.per.ip.overrides = 
sec-broker-one    |     max.incremental.fetch.session.cache.slots = 1000
sec-broker-one    |     message.max.bytes = 1048588
sec-broker-one    |     metric.reporters = []
sec-broker-one    |     metrics.num.samples = 2
sec-broker-one    |     metrics.recording.level = INFO
sec-broker-one    |     metrics.sample.window.ms = 30000
sec-broker-one    |     min.insync.replicas = 1
sec-broker-one    |     num.io.threads = 8
sec-broker-one    |     num.network.threads = 3
sec-broker-one    |     num.partitions = 1
sec-broker-one    |     num.recovery.threads.per.data.dir = 1
sec-broker-one    |     num.replica.alter.log.dirs.threads = null
sec-broker-one    |     num.replica.fetchers = 1
sec-broker-one    |     offset.metadata.max.bytes = 4096
sec-broker-one    |     offsets.commit.required.acks = -1
sec-broker-one    |     offsets.commit.timeout.ms = 5000
sec-broker-one    |     offsets.load.buffer.size = 5242880
sec-broker-one    |     offsets.retention.check.interval.ms = 600000
sec-broker-one    |     offsets.retention.minutes = 10080
sec-broker-one    |     offsets.topic.compression.codec = 0
sec-broker-one    |     offsets.topic.num.partitions = 50
sec-broker-one    |     offsets.topic.replication.factor = 1
sec-broker-one    |     offsets.topic.segment.bytes = 104857600
sec-broker-one    |     password.encoder.cipher.algorithm = AES/CBC/PKCS5Padding
sec-broker-one    |     password.encoder.iterations = 4096
sec-broker-one    |     password.encoder.key.length = 128
sec-broker-one    |     password.encoder.keyfactory.algorithm = null
sec-broker-one    |     password.encoder.old.secret = null
sec-broker-one    |     password.encoder.secret = null
sec-broker-one    |     port = 9092
sec-broker-one    |     principal.builder.class = null
sec-broker-one    |     producer.purgatory.purge.interval.requests = 1000
sec-broker-one    |     queued.max.request.bytes = -1
sec-broker-one    |     queued.max.requests = 500
sec-broker-one    |     quota.consumer.default = 9223372036854775807
sec-broker-one    |     quota.producer.default = 9223372036854775807
sec-broker-one    |     quota.window.num = 11
sec-broker-one    |     quota.window.size.seconds = 1
sec-broker-one    |     replica.fetch.backoff.ms = 1000
sec-broker-one    |     replica.fetch.max.bytes = 1048576
sec-broker-one    |     replica.fetch.min.bytes = 1
sec-broker-one    |     replica.fetch.response.max.bytes = 10485760
sec-broker-one    |     replica.fetch.wait.max.ms = 500
sec-broker-one    |     replica.high.watermark.checkpoint.interval.ms = 5000
sec-broker-one    |     replica.lag.time.max.ms = 30000
sec-broker-one    |     replica.selector.class = null
sec-broker-one    |     replica.socket.receive.buffer.bytes = 65536
sec-broker-one    |     replica.socket.timeout.ms = 30000
sec-broker-one    |     replication.quota.window.num = 11
sec-broker-one    |     replication.quota.window.size.seconds = 1
sec-broker-one    |     request.timeout.ms = 30000
sec-broker-one    |     reserved.broker.max.id = 1000
sec-broker-one    |     sasl.client.callback.handler.class = null
sec-broker-one    |     sasl.enabled.mechanisms = [GSSAPI]
sec-broker-one    |     sasl.jaas.config = null
sec-broker-one    |     sasl.kerberos.kinit.cmd = /usr/bin/kinit
sec-broker-one    |     sasl.kerberos.min.time.before.relogin = 60000
sec-broker-one    |     sasl.kerberos.principal.to.local.rules = [DEFAULT]
sec-broker-one    |     sasl.kerberos.service.name = null
sec-broker-one    |     sasl.kerberos.ticket.renew.jitter = 0.05
sec-broker-one    |     sasl.kerberos.ticket.renew.window.factor = 0.8
sec-broker-one    |     sasl.login.callback.handler.class = null
sec-broker-one    |     sasl.login.class = null
sec-broker-one    |     sasl.login.refresh.buffer.seconds = 300
sec-broker-one    |     sasl.login.refresh.min.period.seconds = 60
sec-broker-one    |     sasl.login.refresh.window.factor = 0.8
sec-broker-one    |     sasl.login.refresh.window.jitter = 0.05
sec-broker-one    |     sasl.mechanism.inter.broker.protocol = GSSAPI
sec-broker-one    |     sasl.server.callback.handler.class = null
sec-broker-one    |     security.inter.broker.protocol = PLAINTEXT
sec-broker-one    |     security.providers = null
sec-broker-one    |     socket.receive.buffer.bytes = 102400
sec-broker-one    |     socket.request.max.bytes = 104857600
sec-broker-one    |     socket.send.buffer.bytes = 102400
sec-broker-one    |     ssl.cipher.suites = []
sec-broker-one    |     ssl.client.auth = required
sec-broker-one    |     ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
sec-broker-one    |     ssl.endpoint.identification.algorithm = 
sec-broker-one    |     ssl.key.password = null
sec-broker-one    |     ssl.keymanager.algorithm = SunX509
sec-broker-one    |     ssl.keystore.location = null
sec-broker-one    |     ssl.keystore.password = null
sec-broker-one    |     ssl.keystore.type = JKS
sec-broker-one    |     ssl.principal.mapping.rules = DEFAULT
sec-broker-one    |     ssl.protocol = TLS
sec-broker-one    |     ssl.provider = null
sec-broker-one    |     ssl.secure.random.implementation = null
sec-broker-one    |     ssl.trustmanager.algorithm = PKIX
sec-broker-one    |     ssl.truststore.location = null
sec-broker-one    |     ssl.truststore.password = null
sec-broker-one    |     ssl.truststore.type = JKS
sec-broker-one    |     transaction.abort.timed.out.transaction.cleanup.interval.ms = 10000
sec-broker-one    |     transaction.max.timeout.ms = 900000
sec-broker-one    |     transaction.remove.expired.transaction.cleanup.interval.ms = 3600000
sec-broker-one    |     transaction.state.log.load.buffer.size = 5242880
sec-broker-one    |     transaction.state.log.min.isr = 2
sec-broker-one    |     transaction.state.log.num.partitions = 50
sec-broker-one    |     transaction.state.log.replication.factor = 3
sec-broker-one    |     transaction.state.log.segment.bytes = 104857600
sec-broker-one    |     transactional.id.expiration.ms = 604800000
sec-broker-one    |     unclean.leader.election.enable = false
sec-broker-one    |     zookeeper.clientCnxnSocket = null
sec-broker-one    |     zookeeper.connect = sec-zookeeper:2181
sec-broker-one    |     zookeeper.connection.timeout.ms = null
sec-broker-one    |     zookeeper.max.in.flight.requests = 10
sec-broker-one    |     zookeeper.session.timeout.ms = 18000
sec-broker-one    |     zookeeper.set.acl = false
sec-broker-one    |     zookeeper.ssl.cipher.suites = null
sec-broker-one    |     zookeeper.ssl.client.enable = false
sec-broker-one    |     zookeeper.ssl.crl.enable = false
sec-broker-one    |     zookeeper.ssl.enabled.protocols = null
sec-broker-one    |     zookeeper.ssl.endpoint.identification.algorithm = HTTPS
sec-broker-one    |     zookeeper.ssl.keystore.location = null
sec-broker-one    |     zookeeper.ssl.keystore.password = null
sec-broker-one    |     zookeeper.ssl.keystore.type = null
sec-broker-one    |     zookeeper.ssl.ocsp.enable = false
sec-broker-one    |     zookeeper.ssl.protocol = TLSv1.2
sec-broker-one    |     zookeeper.ssl.truststore.location = null
sec-broker-one    |     zookeeper.ssl.truststore.password = null
sec-broker-one    |     zookeeper.ssl.truststore.type = null
sec-broker-one    |     zookeeper.sync.time.ms = 2000

Its ignoring almost all of the settings, namely the KAFKA_SSL_KEYSTORE and KAFKA_SSL_TRUSTSTORE variables. I've had a poke around in the container, the configuration values are in kafka.properties so they are getting copied in, but it seems being ignored by the underlying image when running the broker. I've provided a view of the properties file from inside of the broker image:

# cat /etc/kafka/kafka.properties
broker.id=1
ssl.key.credentials=server.password
advertised.listeners=PLAINTEXT://sec-broker-one:9092,CONNECTIONS_FROM_HOST://localhost:19092
offsets.topic.replication.factor=1
ssl.endpoint.identification.algorithm=
ssl.keystore.filename=kafka.server.keystore.jks
zookeeper.connect=sec-zookeeper:2181
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,CONNECTIONS_FROM_HOST:SSL
log.dirs=/var/lib/kafka/data
ssl.truststore.filename=kafka.server.truststore.jks
listeners=PLAINTEXT://0.0.0.0:9092,CONNECTIONS_FROM_HOST://0.0.0.0:19092
ssl.truststore.credentials=server.password
ssl.client.auth=required
log.message.timestamp.type=CreateTime
ssl.keystore.credentials=server.password
log.retention.hours=2232

Some of the properties are coming through fine as you can see from listener.security.protocol.map being set correctly to PLAINTEXT:PLAINTEXT,CONNECTIONS_FROM_HOST:SSL

But not all of them.

CodePudding user response：

Look at the logs, you'll see the variable ends in .location, which requires an absolute path in the container

ssl.*.filename are not valid broker properties, and neither are ssl.*.credentials