Home / Event Gateway
Edit
Report

Enrich Kafka SASL PLAIN connections with Kong Identity principal metadata

Uses: Event Gateway Kong Identity Konnect API
Incompatible with
on-prem
Related Documentation
Event Gateway Documentation
Tags
#event-gateway #kafka
Related Resources
Authenticate Kong Event Gateway connections to Kafka using SASL/PLAIN
Modify Headers policy
Event Gateway expressions language
Backend clusters
Minimum Version
Event Gateway - 1.2.0
TL;DR
  1. Create a Kong Identity directory, principal with metadata, and a custom identity keyed by the SASL username.
  2. Configure a virtual cluster with sasl_plain passthrough authentication and fetch_kong_identity_principal pointing at the directory.
  3. Create a Modify Headers policy with a condition on context.auth.principal.metadata.
  4. Produce and consume a record to see the policy fire.

Prerequisites

Install kafkactl

Install kafkactl. You’ll need it to interact with Kafka clusters.

Kong Konnect

If you don’t have a Konnect account, you can get started quickly with our onboarding wizard.

  1. The following Konnect items are required to complete this tutorial:
    • Personal access token (PAT): Create a new personal access token by opening the Konnect PAT page and selecting Generate Token.

  2. Set the personal access token as an environment variable:

    export KONNECT_TOKEN='YOUR KONNECT TOKEN'

Kong Identity directory

A directory is a regional collection of principals. Create a directory for this tutorial:

DIRECTORY_ID=$(curl -X POST "https://us.api.konghq.com/v2/directories" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN"\
     -H "Content-Type: application/json" \
     --json '{
       "name": "kong-identity-directory",
       "description": "Directory for this tutorial",
       "allow_all_control_planes": true
     }' | jq -r ".id"
)

In this guide, you’ll authenticate a Kafka client to a SASL-secured broker through Event Gateway, look up the connecting principal in a Kong Identity directory by its SASL username, and use the principal’s metadata to drive a Modify Headers policy.

 
flowchart LR
    C[Kafka client]
    subgraph EG [" Event Gateway "]
        VC[sasl_plain passthrough
virtual cluster]
    end
    KI[(Kong Identity
directory)]
    subgraph K [Kafka cluster]
        L["SASL_PLAINTEXT :9082"]
    end
    C -->|SASL/PLAIN
user=john| VC
    VC -.->|lookup by sasl_username| KI
    KI -.->|principal metadata
team=operators| VC
    VC -->|SASL/PLAIN passthrough| L
    VC -->|record with
x-team header| C

Start the secured Kafka cluster

Create the JAAS configuration file that defines the SASL/PLAIN credentials:

cat <<'EOF' > kafka_server_jaas.conf
KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="eventgateway"
    password="eventgateway-secret"
    user_eventgateway="eventgateway-secret"
    user_john="john-secret";
};
EOF

The broker accepts two SASL/PLAIN users: eventgateway (used by Event Gateway itself for broker discovery) and john (used by the Kafka client and matched against Kong Identity).

Create the Docker Compose file:

cat <<'EOF' > docker-compose.yaml
name: kafka_cluster

networks:
  kafka:
    name: kafka_event_gateway

services:
  kafka1:
    image: apache/kafka:4.3.0
    networks:
      - kafka
    container_name: kafka1
    ports:
      - "9094:9094"
    environment:
      KAFKA_NODE_ID: 0
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENERS: INTERNAL://kafka1:9092,CONTROLLER://kafka1:9093,EXTERNAL://0.0.0.0:9094,SASL_PLAINTEXT://0.0.0.0:9082
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka1:9092,EXTERNAL://localhost:9094,SASL_PLAINTEXT://kafka1:9082
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_CONTROLLER_QUORUM_VOTERS: 0@kafka1:9093,1@kafka2:9093,2@kafka3:9093
      KAFKA_CLUSTER_ID: 'abcdefghijklmnopqrstuv'
      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
      KAFKA_SASL_ENABLED_MECHANISMS: PLAIN
      KAFKA_OPTS: -Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf
    volumes:
      - ./kafka_server_jaas.conf:/etc/kafka/kafka_server_jaas.conf

  kafka2:
    image: apache/kafka:4.3.0
    networks:
      - kafka
    container_name: kafka2
    ports:
      - "9095:9095"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENERS: INTERNAL://kafka2:9092,CONTROLLER://kafka2:9093,EXTERNAL://0.0.0.0:9095,SASL_PLAINTEXT://0.0.0.0:9082
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka2:9092,EXTERNAL://localhost:9095,SASL_PLAINTEXT://kafka2:9082
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_CONTROLLER_QUORUM_VOTERS: 0@kafka1:9093,1@kafka2:9093,2@kafka3:9093
      KAFKA_CLUSTER_ID: 'abcdefghijklmnopqrstuv'
      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
      KAFKA_SASL_ENABLED_MECHANISMS: PLAIN
      KAFKA_OPTS: -Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf
    volumes:
      - ./kafka_server_jaas.conf:/etc/kafka/kafka_server_jaas.conf

  kafka3:
    image: apache/kafka:4.3.0
    networks:
      - kafka
    container_name: kafka3
    ports:
      - "9096:9096"
    environment:
      KAFKA_NODE_ID: 2
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENERS: INTERNAL://kafka3:9092,CONTROLLER://kafka3:9093,EXTERNAL://0.0.0.0:9096,SASL_PLAINTEXT://0.0.0.0:9082
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka3:9092,EXTERNAL://localhost:9096,SASL_PLAINTEXT://kafka3:9082
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_CONTROLLER_QUORUM_VOTERS: 0@kafka1:9093,1@kafka2:9093,2@kafka3:9093
      KAFKA_CLUSTER_ID: 'abcdefghijklmnopqrstuv'
      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
      KAFKA_SASL_ENABLED_MECHANISMS: PLAIN
      KAFKA_OPTS: -Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf
    volumes:
      - ./kafka_server_jaas.conf:/etc/kafka/kafka_server_jaas.conf

EOF

The broker exposes a SASL_PLAINTEXT listener on port 9082 in the Docker network for Event Gateway connections, and a PLAINTEXT listener on ports 9094/9095/9096 for direct local access.

Start the cluster:

docker compose up -d

Create an Event Gateway control plane and data plane

Run the quickstart script to provision a local data plane and configure your environment:

curl -Ls https://get.konghq.com/event-gateway | bash -s -- -k $KONNECT_TOKEN -N kafka_event_gateway

Copy the exported variable into your terminal:

export EVENT_GATEWAY_ID=your-gateway-id

This quickstart script is meant for demo purposes only, therefore it runs locally with most default parameters and a small number of exposed ports. If you want to run Kong Gateway as a part of a production-ready platform, set up your control plane and data planes through the Konnect UI, or using Terraform.

Create a principal with team metadata

Create a principal in the directory and attach the team metadata. The Modify Headers policy will read this value at request time:

PRINCIPAL_ID=$(curl -X POST "https://us.api.konghq.com/v2/directories/$DIRECTORY_ID/principals" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "display_name": "john",
       "description": "Principal that maps to the john SASL user",
       "metadata": {
         "team": "operators"
       }
     }' | jq -r ".id"
)

Create a custom identity for the SASL username

Create a custom identity that links the principal to the SASL username sent by the Kafka client. Event Gateway will match the connecting username against the sasl_username key:

curl -X POST "https://us.api.konghq.com/v2/directories/$DIRECTORY_ID/principals/$PRINCIPAL_ID/identities" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "type": "custom",
       "key": "sasl_username",
       "value": "john"
     }'

Create the backend cluster

Create a backend cluster configured with the eventgateway SASL/PLAIN user. Event Gateway uses these credentials for its own connection to the broker. Client connections pass through this configuration unchanged because of the virtual cluster’s passthrough mediation, which you’ll configure in the next step:

BACKEND_CLUSTER_ID=$(curl -X POST "https://us.api.konghq.com/v1/event-gateways/$EVENT_GATEWAY_ID/backend-clusters" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "name": "backend_cluster",
       "bootstrap_servers": [
         "kafka1:9082",
         "kafka2:9082",
         "kafka3:9082"
       ],
       "authentication": {
         "type": "sasl_plain",
         "username": "eventgateway",
         "password": "eventgateway-secret"
       },
       "tls": {
         "enabled": false
       }
     }' | jq -r ".id"
)

Create a virtual cluster

Create a virtual cluster that accepts SASL/PLAIN connections, forwards them unchanged to the broker, and asks Event Gateway to fetch the principal from the Kong Identity directory by matching the SASL username against the sasl_username key:

VIRTUAL_CLUSTER_ID=$(curl -X POST "https://us.api.konghq.com/v1/event-gateways/$EVENT_GATEWAY_ID/virtual-clusters" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "name": "identity_vc",
       "destination": {
         "id": "'$BACKEND_CLUSTER_ID'"
       },
       "dns_label": "identity-vc",
       "acl_mode": "passthrough",
       "authentication": [
         {
           "type": "sasl_plain",
           "mediation": "passthrough",
           "fetch_kong_identity_principal": {
             "directory": "kong-identity-directory",
             "fetch_by": {
               "key": "sasl_username"
             },
             "failure_mode": "error"
           }
         }
       ]
     }' | jq -r ".id"
)

The fetch_kong_identity_principal block tells Event Gateway to use the SASL username (in this case, john) as the lookup value against identities of key sasl_username in the directory. When a match is found, the parent principal’s metadata is attached to context.auth.principal.metadata for the lifetime of the connection.

Create a listener

Run the following command to create a new listener:

LISTENER_ID=$(curl -X POST "https://us.api.konghq.com/v1/event-gateways/$EVENT_GATEWAY_ID/listeners" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "name": "identity_listener",
       "addresses": [
         "0.0.0.0"
       ],
       "ports": [
         "19092-19095"
       ]
     }' | jq -r ".id"
)

Create a listener policy

Add a Forward to Virtual Cluster policy that routes the listener to the virtual cluster:

curl -X POST "https://us.api.konghq.com/v1/event-gateways/$EVENT_GATEWAY_ID/listeners/$LISTENER_ID/policies" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "type": "forward_to_virtual_cluster",
       "name": "forward_to_identity_vc",
       "config": {
         "type": "port_mapping",
         "advertised_host": "localhost",
         "destination": {
           "id": "'$VIRTUAL_CLUSTER_ID'"
         }
       }
     }'

Create the Modify Headers policy

Add a Modify Headers policy that sets the x-team header on consumed records only when the principal’s team metadata equals operators:

curl -X POST "https://us.api.konghq.com/v1/event-gateways/$EVENT_GATEWAY_ID/virtual-clusters/$VIRTUAL_CLUSTER_ID/consume-policies" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "type": "modify_headers",
       "name": "tag-operators-team",
       "condition": "context.auth.principal.metadata.team == \"operators\"",
       "config": {
         "actions": [
           {
             "op": "set",
             "key": "x-team",
             "value": "operators"
           }
         ]
       }
     }'

Configure kafkactl

Create a kafkactl.yaml config file with a direct context that talks to the broker’s PLAINTEXT listener, and a vc context that connects to the virtual cluster using SASL/PLAIN:

cat <<EOF > kafkactl.yaml
contexts:
  direct:
    brokers:
      - localhost:9094
      - localhost:9095
      - localhost:9096
  vc:
    brokers:
      - localhost:19092
    sasl:
      enabled: true
      username: john
      password: john-secret
EOF

Create a topic

Create the orders topic using the direct context:

kafkactl -C kafkactl.yaml --context direct create topic orders

Validate

Produce a record through the virtual cluster:

kafkactl -C kafkactl.yaml --context vc produce orders --value="test-message"

Consume the record back through the virtual cluster with --print-headers so you can see the header added by the Modify Headers policy:

kafkactl -C kafkactl.yaml --context vc consume orders --print-headers --from-beginning --exit

The output should contain the x-team header:

x-team:operators#test-message

Event Gateway authenticated the client with the broker by passing the SASL/PLAIN credentials straight through, looked up the john SASL username in the Kong Identity directory, attached the principal’s metadata to the connection, and applied the Modify Headers policy because context.auth.principal.metadata.team was operators.

The same principal lookup strategy can be used with all other authentication methods (SASL/SCRAM, SASL/OAUTHBEARER, client certificates).

Cleanup

Clean up Kong Event Gateway resources

When you’re done experimenting with this example, clean up the resources:

  1. If you created a new Event Gateway control plane and want to conserve your free trial credits or avoid unnecessary charges, delete the new control plane used in this tutorial.

  2. Stop and remove the containers:

    docker-compose down

This will stop all services and remove the containers, but preserve your configuration files for future use.

Something wrong?
Report an Issue | Edit this Page
View all Kong documentation

Help us make these docs great!

Kong Developer docs are open source. If you find these useful and want to make them better, contribute today!
Contribute

Still need help?

Ask in our Slack Community
Contact Support