Security & Production

ไอเดียในหนึ่งประโยค

security ของ Kafka คือสามชั้นที่พิสูจน์แยกกันได้จริง — authorization (ACL ซึ่งไม่ทำอะไรเลยจนกว่าจะเปิด authorizer), authentication (SASL หรือ mTLS ซึ่งตัดสินว่า principal คือใครกันแน่), และ encryption (TLS) — การข้ามขั้นตอนแรกไป อย่างที่ tutorial จำนวนมากทำ จะได้คำสั่งที่ดูเหมือนควรทำงานได้แต่กลับ error ทันที

พิสูจน์: `kafka-acls.sh` ตอนไม่มี authorizer

Kafka มาโดยไม่มี authorizer ติดมา by default รันคำสั่งเดียวกับที่ tutorial จำนวนมากเปิดด้วย บน broker ที่ยังไม่เปิด authorizer

$ kafka-acls.sh --bootstrap-server localhost:9092 --list
Error while executing ACL command: org.apache.kafka.common.errors.SecurityDisabledException: No Authorizer is configured on the broker
java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.SecurityDisabledException: No Authorizer is configured on the broker

นี่ไม่ใช่ permission error — เป็น kafka-acls.sh ที่บอกคุณอย่างถูกต้องว่าไม่มีระบบ ACL ใด ๆ รันอยู่เลยให้ query วิธีแก้คือ broker config ที่ต้องตั้งก่อน broker สตาร์ท

# section: broker
authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer

StandardAuthorizer คือ authorizer แบบ KRaft-native (AclAuthorizer ยุค ZooKeeper หายไปพร้อมกับ ZooKeeper แล้ว) จับคู่กับ super.users (มีเอกสารอยู่ใต้ configuration ของ authorization/ACL ของ Kafka ไม่ใช่ตาราง broker-configs ทั่วไปด้านบน) — list แบบ principalType:name คั่นด้วย comma ที่ข้ามการเช็ก ACL ทุกอัน cluster จริงทุกอันต้องมี super user อย่างน้อยหนึ่งตัว ไม่งั้น authorizer ที่ล็อกแน่นอาจล็อก admin ของตัวเองออกได้

super.users=User:ANONYMOUS

🌊 gotcha จริงที่บทนี้เจอตอนตั้ง super.users

broker ของบทนี้เองใช้ PLAINTEXT สำหรับ listener ภายใน/admin ซึ่งไม่มี authentication เลย — ทุก connection ผ่านมันจะได้ principal User:ANONYMOUS การใส่ User:ANONYMOUS เข้า super.users เป็นทางลัดเพื่อการสอนโดยตั้งใจ เพื่อให้คำสั่ง admin ด้านล่าง (รันผ่าน PLAINTEXT listener เดียวกัน) ยังทำงานได้หลังเปิด authorizer แล้ว cluster production จริงไม่ควรปล่อยให้ listener ที่ไม่มี authentication เป็น super user ตลอดไป — มันแปลว่าใครก็ตามที่เข้าถึง listener นั้นได้คือ admin เต็มรูปแบบ

พิสูจน์: เปิด authorizer แล้ว แต่ยังไม่มี ACL — produce ที่ถูกปฏิเสธจริง

หลังเปิด authorizer และเพิ่ม SASL/SCRAM listener (setup เต็มด้านล่าง) client ที่ authenticate เป็น order-service — principal จริง ไม่ใช่ super user — พยายาม produce เข้า topic ที่ไม่เคยได้รับสิทธิ์เลย

$ kafka-console-producer.sh --bootstrap-server localhost:39094 --topic secure-orders \
    --command-config order-service-client.properties
[...] WARN [Producer clientId=console-producer] The metadata response from the cluster reported a recoverable issue with correlation id 1 : {secure-orders=TOPIC_AUTHORIZATION_FAILED} (org.apache.kafka.clients.NetworkClient)
[...] ERROR [Producer clientId=console-producer] Topic authorization failed for topics [secure-orders] (org.apache.kafka.clients.Metadata)
org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [secure-orders]

TopicAuthorizationException จริงแบบ verbatim — authentication ผ่าน (SCRAM ตรวจ password ผ่านแล้ว พิสูจน์ต่อด้านล่าง) แต่ authorization ไม่ผ่าน kafka-acls.sh --add คือวิธีแก้

$ kafka-acls.sh --bootstrap-server localhost:9092 \
    --add --allow-principal User:order-service \
    --operation Write --topic secure-orders
Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=secure-orders, patternType=LITERAL)`:
   (principal=User:order-service, host=*, operation=WRITE, permissionType=ALLOW)

$ kafka-console-producer.sh --bootstrap-server localhost:39094 --topic secure-orders \
    --command-config order-service-client.properties
(exits 0, no output -- the record went through)

sequenceDiagram
  participant C as order-service (authenticated, not super)
  participant B as Broker
  C->>B: Produce to secure-orders (no ACL granted)
  B-->>C: TOPIC_AUTHORIZATION_FAILED -> real TopicAuthorizationException
  Note over B: kafka-acls.sh --add --allow-principal User:order-service --operation Write --topic secure-orders
  C->>B: Produce to secure-orders (retry, same credentials)
  B-->>C: accepted -- exit 0

Authorization พิสูจน์ว่าผิดก่อน แล้วถูก ด้วย principal เดียวกัน

bug จริงของ Docker image สองตัวที่เจอตอนตั้งค่า

การสร้าง broker SASL/SCRAM ของบทนี้เองเจอ failure จริงสองตัวที่ไม่มีเอกสารบอกจนกว่าจะเจอเอง ใน official image ทั้งคู่ — คุ้มค่าที่จะรู้ไว้ เพราะทั้งคู่ทำให้ container crash ตอนสตาร์ทโดยไม่เกี่ยวอะไรกับ Kafka config ที่ผิดเลย

🌊 Bug 1: การเปิด listener SASL_* ตัวไหนก็ตามต้องการ KAFKA_OPTS แบบเงียบ ๆ

script /etc/kafka/docker/configure ของ image เองมี branch ที่รันเมื่อ KAFKA_ADVERTISED_LISTENERS ตรงกับ protocol SASL_* — และ branch นั้นเรียก ensure KAFKA_OPTS ซึ่งต้องการให้ KAFKA_OPTS ถูกตั้งค่าอะไรสักอย่าง ด้วยเหตุผลที่ไม่เกี่ยวกับว่า SASL/SCRAM ทำงานได้จริงหรือไม่เลย ถ้าไม่ตั้งไว้ container จะ crash ที่ชั้น shell script ก่อนที่ JVM จะสตาร์ทด้วยซ้ำ ด้วย bash error ดิบ ๆ แทนที่จะเป็น error ของ Kafka: /etc/kafka/docker/configure: line 18: !1: unbound variable วิธีแก้คือใส่ค่า KAFKA_OPTS อะไรก็ได้ที่ไม่ว่าง เช่น KAFKA_OPTS=-Dkafka.docker.compose=security-lesson

🌊 Bug 2: env var ของ JAAS แบบต่อ listener ต้องขีดล่างซ้อนสอง/สามตัว

ตัวแปลง KAFKA_* env-var เป็น property ของ image ถือว่าขีดล่างตัวเดียวคือตัวแบ่ง dot property ที่ต้องการ override JAAS แบบต่อ listener ต่อ mechanism คือ listener.name.sasl_plaintext_host.scram-sha-256.sasl.jaas.config — และทั้งขีดล่างของชื่อ listener เอง (sasl_plaintext_host) และขีดกลางของ mechanism (scram-sha-256) เป็นตัวอักษรจริงในตัว property นั้น ไม่ใช่ตัวแบ่ง ถ้าตั้งผิด (KAFKA_LISTENER_NAME_SASL_PLAINTEXT_HOST_SCRAM_SHA_256_SASL_JAAS_CONFIG แบบขีดล่างเดี่ยวล้วน) broker จะ crash ด้วย IllegalArgumentException: Could not find a 'KafkaServer' or 'sasl_plaintext_host.KafkaServer' entry in the JAAS configuration จริง กฎจริงคือ: ขีดล่างจริงในหนึ่ง segment ต้องซ้อนสองตัว (__); ขีดกลางจริงต้องซ้อนสามตัว (___); มีแค่ขีดล่างเดี่ยวจริง ๆ เท่านั้นที่แปลว่า “ใส่ dot ตรงนี้” — ได้เป็น KAFKA_LISTENER_NAME_SASL__PLAINTEXT__HOST_SCRAM___SHA___256_SASL_JAAS_CONFIG ซึ่งสตาร์ทสะอาดทันที

SASL/SCRAM ใน KRaft ครบวงจร

มีสองทางในการตั้ง SCRAM credential ใน KRaft (ไม่มี ZooKeeper เลยไม่มี flag --zookeeper แยกสำหรับเรื่องนี้ด้วย): ตอน format หรือแบบ dynamic กับ cluster ที่รันอยู่ flag ตอน format ยืนยันจาก help text จริงของ tool เอง

/opt/kafka/bin/kafka-storage.sh format --help

  --add-scram ADD_SCRAM, -S ADD_SCRAM
                         A SCRAM_CREDENTIAL to add to the __cluster_metadata log e.g.
                         'SCRAM-SHA-256=[name=alice,password=alice-secret]'
                         'SCRAM-SHA-512=[name=alice,iterations=8192,salt="N3E=",saltedpassword="YCE="]'

บทนี้ใช้ทางที่สอง — เพิ่ม user เข้า cluster ที่รันอยู่แล้ว ซึ่งเป็นสิ่งที่ workflow ปฏิบัติงานจริงส่วนใหญ่ทำ (เพิ่ม/หมุนเวียน credential โดยไม่ต้อง reformat อะไรเลย)

# Self-contained: creates its own SCRAM user, proves it, then deletes it.
/opt/kafka/bin/kafka-configs.sh --bootstrap-server "$BOOTSTRAP" \
  --alter --entity-type users --entity-name lesson-scram-demo \
  --add-config 'SCRAM-SHA-256=[password=demo-secret-pw]'

/opt/kafka/bin/kafka-configs.sh --bootstrap-server "$BOOTSTRAP" \
  --describe --entity-type users --entity-name lesson-scram-demo

/opt/kafka/bin/kafka-configs.sh --bootstrap-server "$BOOTSTRAP" \
  --alter --entity-type users --entity-name lesson-scram-demo \
  --delete-config 'SCRAM-SHA-256'

Completed updating config for user lesson-scram-demo.
SCRAM credential configs for user-principal 'lesson-scram-demo' are SCRAM-SHA-256=iterations=4096
Completed updating config for user lesson-scram-demo.

ตัว credential เอง — ไม่ใช่ plaintext password — คือสิ่งที่ถูกเก็บ: --describe แสดง iterations=4096 ไม่เคยแสดง demo-secret-pw listener ต้องเปิดใช้ SCRAM จริงด้วย

# section: broker
sasl.enabled.mechanisms=SCRAM-SHA-256
security.inter.broker.protocol=PLAINTEXT

พิสูจน์: client จริงที่ authenticate ผ่าน — และ client จริงที่ล้มเหลว

KafkaJS producer จริง รันจาก host เข้า SASL_PLAINTEXT listener ด้วย credential ที่ถูกต้อง

const kafka = new Kafka({
  clientId: 'security-lesson-probe',
  brokers: ['localhost:39094'],
  sasl: { mechanism: 'scram-sha-256', username: 'order-service', password: 'order-secret-pw' },
});
const producer = kafka.producer();
await producer.connect();
await producer.send({ topic: 'secure-orders', messages: [{ value: 'hello from KafkaJS over SASL_PLAINTEXT' }] });

SASL/SCRAM connect OK as order-service
send OK

client ตัวเดียวกันเป๊ะ ใส่ password ผิดอย่างเดียว

{"level":"ERROR", ..., "message":"[SaslAuthenticator-SCRAM-SHA-256] SASL SCRAM SHA256 authentication failed: Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-256", ...}
connect FAILED: KafkaJSSASLAuthenticationError - SASL SCRAM SHA256 authentication failed: Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-256

และฝั่ง Java tooling ของความผิดพลาดเดียวกันเป๊ะ เพื่อเทียบกัน — broker เดียวกัน password ผิดเหมือนกัน ผ่าน kafka-console-producer.sh แทน

[...] ERROR [Producer clientId=console-producer] Connection to node -1 (localhost/127.0.0.1:39094) failed authentication due to: Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-256 (org.apache.kafka.clients.NetworkClient)
org.apache.kafka.common.errors.SaslAuthenticationException: Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-256

client ทั้งคู่โดนเช็กฝั่ง broker อันเดียวกันเป๊ะ ได้เหตุผลที่มนุษย์อ่านได้เหมือนกันเป๊ะ — ต่างกันแค่ชื่อ exception class ตาม client (KafkaJSSASLAuthenticationError เทียบ SaslAuthenticationException) เพราะ challenge-response ของ SCRAM เกิดขึ้นบน wire ไม่ใช่ข้างในของ client library ตัวไหนเลย

TLS: รูปร่างของมัน และ config key จริง

บทนี้ไม่ได้ generate certificate ครบวงจร — config key ฝั่ง broker คือส่วนที่สำคัญ และมีเอกสารรองรับ ไม่ใช่ตำนาน ยืนยันกับ broker config reference ทางการ (kafka.apache.org/43/configuration/ ส่วน broker ที่ดึงมาสำหรับคอร์สนี้)

# section: broker
listeners=SSL://:9093
ssl.keystore.location=/etc/kafka/secrets/broker.keystore.jks
ssl.keystore.password=changeit
ssl.key.password=changeit
ssl.truststore.location=/etc/kafka/secrets/broker.truststore.jks
ssl.truststore.password=changeit
ssl.client.auth=required
ssl.endpoint.identification.algorithm=https

แต่ละ broker ต้องมีทั้งสอง keystore (identity ของตัวเอง — certificate ที่มันนำเสนอ) และ truststore (ใครที่มันยอมเชื่อถือ — ปกติคือ CA ที่ sign certificate ของทุก broker และ client ใน private cluster) ssl.client.auth=required เปิด TLS แบบ mutual: broker เรียกร้องและตรวจ certificate จาก client ด้วย ซึ่งเป็นโหมด mTLS authentication ที่กล่าวถึงด้านล่าง — TLS อย่างเดียวพิสูจน์แค่ identity ของ broker ให้ client เห็น ไม่ใช่ในทางกลับกัน ssl.endpoint.identification.algorithm=https เปิด hostname verification กับ SAN ของ certificate ซึ่งป้องกัน man-in-the-middle แบบง่าย ๆ แม้ attacker จะมี certificate ที่ valid จริง แต่ไม่ใช่สำหรับ hostname ที่กำลังเชื่อมต่อ — การไม่ตั้งค่านี้คือ misconfiguration จริงที่พบบ่อย

checklist ความพร้อมสำหรับ production

security เป็นคอลัมน์เดียว durability กับ observability เป็นคอลัมน์อื่นที่ครอบคลุมใน Monitoring & Consumer Lag และ Performance & Tuning ก่อน cluster จะรับ traffic จริง

authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer พร้อม super.users ที่จำกัดเฉพาะ principal ของ admin จริง ๆ — พิสูจน์ด้านบนว่าต้องมีก่อน ACL จะมีความหมายอะไรเลย
TLS ทุกที่ (ssl.client.auth=required สำหรับ mTLS หรือ SASL ผ่าน SASL_SSL สำหรับ access ที่ authenticated และ encrypted) — listener แบบ plaintext ไม่ควรเข้าถึงได้จากนอก trusted network boundary
ACL แบบ least-privilege ต่อ principal — พิสูจน์ด้านบนด้วย order-service ที่ได้แค่ Write บน secure-orders เท่านั้น ไม่กว้างกว่านั้น
Replication factor >= 3 บน topic สำคัญพร้อม min.insync.replicas=2 และ producer ใช้ acks=all — combination ที่ durable ซึ่งบท Acks & Durability ของคอร์สนี้พิสูจน์ไว้
Monitoring และ alerting บน consumer lag, under-replicated partition, และ request latency — หัวข้อของบทถัดไป
Quota เพื่อไม่ให้ client ตัวเดียวแย่ง bandwidth หรือ request capacity ของ cluster ทั้งหมด

🌊 สรุปในหนึ่งบรรทัด

kafka-acls.sh ไม่ทำอะไรเลยจนกว่า authorizer.class.name จะถูกตั้ง — พิสูจน์ด้านบนด้วย SecurityDisabledException จริง พอตั้งแล้ว ACL ถูกบังคับใช้จริง: principal ที่ไม่มีสิทธิ์ได้ TopicAuthorizationException จริง และ kafka-acls.sh --add คือสิ่งที่แก้ได้จริง พิสูจน์ครบวงจรบน principal เดียวกัน SASL/SCRAM ใน KRaft เก็บ credential metadata จริง (iterations=4096 ไม่เคยเป็น plaintext password) ผ่าน kafka-configs.sh --alter --entity-type users และ password ผิดได้การปฏิเสธฝั่ง broker เหมือนกันไม่ว่า client จะเป็น KafkaJS หรือ Java console tools bug จริงของ Docker image สองตัว (ความต้องการ KAFKA_OPTS แบบเงียบ ๆ และการ escape ขีดล่าง/ขีดกลางใน JAAS env var แบบต่อ listener) กินเวลา debug มากกว่า security config จริงเสียอีก TLS ต้องมีทั้ง keystore และ truststore ต่อ broker และ ssl.client.auth=required บวก ssl.endpoint.identification.algorithm=https คือสิ่งที่เปลี่ยน “encrypted” ให้กลายเป็น “authenticated ร่วมกันและ verify hostname แล้ว”

อ่านโค้ดที่ AI เขียน

🐛 หา bug ในโค้ดนี้

#!/usr/bin/env bash
# AI comment: "Lock down the orders topic to only the order-service producer"
set -euo pipefail

kafka-topics.sh --bootstrap-server localhost:9092 \
--create --topic orders --partitions 3 --replication-factor 3

kafka-acls.sh --bootstrap-server localhost:9092 \
--add --allow-principal User:order-service \
--operation Write --topic orders

เฉลย

script นี้คือ runbook “ล็อก topic” ที่ดูสมเหตุสมผลเต็มที่ — สร้างมันขึ้นมา แล้วให้สิทธิ์เฉพาะ principal เดียวที่ควรเขียนเข้าไปได้ ไม่กว้างกว่านั้น มันมีแนวโน้มสูงมากที่จะพังบน cluster จริงด้วย SecurityDisabledException เดียวกับที่พิสูจน์ไว้ต้นบทนี้: ไม่มีอะไรใน script นี้ (หรือที่ไหนก่อนหน้านี้ ถ้านี่คือคำสั่งแรกที่เกี่ยวกับ security ที่ทีมรัน) เคยตั้ง authorizer.class.name เลย kafka-acls.sh --add บน broker ที่ไม่มี authorizer จะไม่ no-op เงียบ ๆ — มัน error ออกมาทันที set -euo pipefail ฆ่า script ตรงนั้นเลย และ topic ที่บรรทัด 6 เพิ่งสร้างถูกทิ้งไว้โดยไม่มี ACL enforcement เลย ไม่ใช่สถานะที่ล็อกแน่นแบบที่ comment สัญญาไว้ วิธีแก้ไม่ใช่ flag อื่นของ kafka-acls.sh — แต่คือ authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer ระดับ broker (บวก super.users สำหรับใครก็ตามที่รัน script นี้) ที่ใส่แล้ว restart broker ก่อน ที่จะคาดหวังให้คำสั่ง ACL ใด ๆ ทำอะไรได้จริง

Mental Model

มอง security ของ Kafka เป็นสวิตช์ yes/no อิสระสามตัว เช็กตามลำดับคงที่สำหรับทุก request: connection encrypted หรือไม่ (TLS หรือไม่ก็ได้ — เป็นอิสระจากอีกสองตัว), broker รู้ ว่าใครกำลังถาม หรือไม่ (authentication — SASL, mTLS, หรือ “ทุกคนคือ ANONYMOUS” by default), และ identity นั้นได้รับอนุญาตให้ทำสิ่งนี้เฉพาะเจาะจงหรือไม่ (authorization — ACL ซึ่งบทนี้พิสูจน์แล้วว่าไม่มีอยู่จริงในฐานะ concept เลยจนกว่า authorizer.class.name จะถูกตั้ง) cluster production ต้องการทั้งสามตัว tutorial และ script ที่ AI เขียนจำนวนมากแสดงแค่ตัวที่สาม โดยสมมติเงียบ ๆ ว่าสองตัวแรกถูกจัดการแล้ว — ซึ่งเป็นช่องว่างเดียวกันเป๊ะกับที่ SpotTheBug ของบทนี้เดินเข้าไป

Trade-offs

User:ANONYMOUS เป็น super user บน listener ภายใน — สะดวกสำหรับ admin scripting บน listener ที่ไม่มีอะไรที่ไม่น่าเชื่อถือเข้าถึงได้ (พิสูจน์เป็น pattern ของบทนี้เอง) และเป็นช่องโหว่ security จริงทันทีที่ listener นั้นเข้าถึงได้จากที่ที่ไม่ควร
SASL/SCRAM เทียบกับ mTLS — SCRAM เป็นแบบ self-contained (credential อยู่ใน metadata log ของ cluster เอง พิสูจน์ด้านบน ไม่ต้องมี CA ภายนอก) และหมุนเวียนต่อ user ได้ง่าย; mTLS ผูก identity เข้ากับ certificate ซึ่งรวมศูนย์ trust ไว้ที่ CA แต่หลีกเลี่ยงการส่ง password ไปที่ไหนเลย รวมถึงเข้าไปใน JAAS config string
การตั้ง SCRAM user แบบ dynamic (kafka-configs.sh --alter) เทียบกับตอน format (kafka-storage.sh format --add-scram) — dynamic คือสิ่งที่ operation จริงใช้ (เพิ่ม/หมุนเวียน user โดยไม่ต้องแตะ storage ของทุก broker) พิสูจน์ครบวงจรด้านบน; format-time เหมาะสำหรับ bootstrap credential ของ admin ตัวแรกสุดก่อนที่ broker ไหนจะเคยสตาร์ทเท่านั้นจริง ๆ
ssl.endpoint.identification.algorithm=https เปิดเทียบกับปิด — เปิด (default ที่ปลอดภัยกว่า) ปฏิเสธ certificate ที่ valid แต่ผิด hostname; ปิดยอมรับมัน ซึ่งบางครั้งจำเป็นสำหรับ setup แบบ service-mesh ภายในที่ naming ไม่มาตรฐาน แต่เปิดช่องว่าง man-in-the-middle ที่ hostname verification มีไว้เพื่อปิดกลับมาอีกครั้ง

ความเข้าใจผิดที่พบบ่อย

“kafka-acls.sh --add ทำงานได้บน broker ไหนก็ได้ คุณต้องมี authorizer ก็ต่อเมื่อต้องการให้ ACL ถูก enforce จริงเท่านั้น” ไม่ — พิสูจน์ด้านบน: ถ้าไม่ตั้ง authorizer.class.name kafka-acls.sh จะแม้แต่ list ACL ก็ไม่ได้เลย SecurityDisabledException จริง ไม่ใช่ no-op แบบนุ่มนวล
“TopicAuthorizationException แปลว่า credential ของ client ผิด” ไม่ — พิสูจน์ด้านบน: order-service authenticate สำเร็จ (SCRAM ตรวจผ่าน) แต่ยังถูกปฏิเสธอยู่ดี เพราะ authentication กับ authorization เป็นการเช็กสองอย่างแยกกัน; กรณี password ผิดที่พิสูจน์ด้านบนด้วยเช่นกัน ให้ exception คนละตัวเลย (SaslAuthenticationException / KafkaJSSASLAuthenticationError) ที่จุดต่างกันของ connection
“SASL/SCRAM ต้องการ ZooKeeper หรือ credential store แยกต่างหาก” ไม่ — พิสูจน์ด้านบน: kafka-configs.sh --alter --entity-type users เขียน credential ตรงเข้า __cluster_metadata log ของ KRaft บน cluster ที่รันอยู่ ไม่มี ZooKeeper ไม่มี store ภายนอก
“TLS กับ authentication คือสิ่งเดียวกัน — เปิด SSL:// แปลว่า client ถูก authenticate แล้ว” ไม่ — พิสูจน์ใน config ด้านบน: SSL:// เปล่า ๆ พิสูจน์แค่ identity ของ broker ให้ client เห็นเท่านั้น client ไม่นำเสนอ identity อะไรเลยเว้นแต่ ssl.client.auth=required (mTLS) หรือ SASL mechanism ถูกวางซ้อนผ่าน SASL_SSL