Home / Event Gateway
Edit
Report

Enrich Kafka OAuth 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
Set up Kong Event Gateway with Kong Identity OAuth
Enrich Kafka SASL PLAIN connections with Kong Identity principal metadata
Modify Headers policy
Event Gateway expressions language
Minimum Version
Event Gateway - 1.2.0
TL;DR
  1. Create a Kong Identity auth server, scope, and client.
  2. Create a Kong Identity directory, principal with metadata, and an oidc identity that matches the JWT’s iss and sub.
  3. Configure a virtual cluster with oauth_bearer authentication and fetch_kong_identity_principal pointing at the directory.
  4. Create a Modify Headers policy with a condition on context.auth.principal.metadata.
  5. Produce and consume a record to see the policy fire.

Prerequisites

Install kafkactl

Install kafkactl. You’ll need it to interact with Kafka clusters. Version >= 5.17.0 is needed to support script driven OAuth token generation.

Start a local Kafka cluster

Start a Docker Compose cluster with multiple Kafka services.

First, we need to create a docker-compose.yaml file. This file will define the services we want to run in our local environment:

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
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka1:9092,EXTERNAL://localhost:9094
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL: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

  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
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka2:9092,EXTERNAL://localhost:9095
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL: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

  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
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka3:9092,EXTERNAL://localhost:9096
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL: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

  schema-registry:
    image: confluentinc/cp-schema-registry:8.2.1
    networks:
      - kafka
    container_name: schema-registry
    depends_on:
      - kafka1
      - kafka2
      - kafka3
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka1:9092,kafka2:9092,kafka3:9092

EOF

Now, let’s start the local setup:

docker compose up -d

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 Event Gateway running

Run the quickstart script to automatically provision a demo Kong Gateway control plane and data plane, and configure your environment:

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

This sets up an Kong Gateway control plane named event-gateway-quickstart, provisions a local data plane, and prints out the following environment variable export:

export EVENT_GATEWAY_ID=your-gateway-id

Copy and paste the command with your Event Gateway ID into your terminal to configure your session.

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.

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 Event Gateway with a JWT issued by a Kong Identity auth server, look up the connecting principal in a Kong Identity directory by the token’s issuer and subject, and use the principal’s metadata to drive a Modify Headers policy.

For oauth_bearer authentication, Event Gateway always looks up the Kong Identity identity by matching the JWT’s iss and sub claims against an oidc identity in the directory. No extra lookup-key configuration is needed.

 
flowchart LR
    C[Kafka client]
    subgraph EG [" Event Gateway "]
        VC[oauth_bearer
virtual cluster]
    end
    KI[(Kong Identity
directory)]
    subgraph K [Kafka cluster]
        L["PLAINTEXT :9092"]
    end
    C -->|SASL/OAUTHBEARER
JWT| VC
    VC -.->|lookup by iss + sub| KI
    KI -.->|principal metadata
team=operators| VC
    VC -->|forward request| L
    VC -->|record with
x-team header| C

Create an auth server in Kong Identity

Create an auth server using the /v1/auth-servers endpoint:

_response=$(curl -X POST "https://us.api.konghq.com/v1/auth-servers" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN"\
     -H "Content-Type: application/json" \
     --json '{
       "name": "Event Gateway Auth",
       "audience": "http://event-gateway",
       "description": "Auth server for Event Gateway"
     }')

Export the env variables:

export AUTH_SERVER_ID=$(echo "$_response" | jq -r ".id")
export ISSUER_URL=$(echo "$_response" | jq -r ".issuer")

Configure the auth server with scopes

Configure a scope using the /v1/auth-servers/$AUTH_SERVER_ID/scopes endpoint:

SCOPE_ID=$(curl -X POST "https://us.api.konghq.com/v1/auth-servers/$AUTH_SERVER_ID/scopes" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN"\
     -H "Content-Type: application/json" \
     --json '{
       "name": "kafka",
       "description": "Scope for Kafka access",
       "default": false,
       "include_in_metadata": false,
       "enabled": true
     }' | jq -r ".id"
)

Create a client in the auth server

The client is the machine-to-machine credential. Konnect autogenerates the client ID and secret. Configure the client using the /v1/auth-servers/$AUTH_SERVER_ID/clients endpoint:

_response=$(curl -X POST "https://us.api.konghq.com/v1/auth-servers/$AUTH_SERVER_ID/clients" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN"\
     -H "Content-Type: application/json" \
     --json '{
       "name": "kafka-client",
       "grant_types": [
         "client_credentials"
       ],
       "allow_all_scopes": false,
       "allow_scopes": [
         "'$SCOPE_ID'"
       ],
       "access_token_duration": 3600,
       "id_token_duration": 3600,
       "response_types": [
         "id_token",
         "token"
       ]
     }')

Export the env variables:

export CLIENT_SECRET=$(echo "$_response" | jq -r ".client_secret")
export CLIENT_ID=$(echo "$_response" | jq -r ".id")

Create a principal with team metadata

Create a principal in the directory you created in the prerequisites 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 for the kafka-client OAuth client",
       "metadata": {
         "team": "operators"
       }
     }' | jq -r ".id"
)

Create an OIDC identity for the JWT subject

Create an oidc identity that links the principal to the JWT issued by the auth server. Event Gateway will match the JWT’s iss against issuer and the JWT’s sub against the configured claim:

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": "oidc",
       "issuer": "'$ISSUER_URL'",
       "claim": {
         "name": "sub",
         "value": "'$CLIENT_ID'"
       }
     }'

For the client_credentials grant, Kong Identity sets the JWT sub claim to the client ID, so the identity’s claim.value is the CLIENT_ID captured earlier.

Connect the event-gateway-quickstart to the same network as Kafka

Configure the event-gateway-quickstart container you created in the prerequisites to use the same network as your Kafka cluster:

docker network connect kafka_event_gateway event-gateway-quickstart

This allows the two containers to communicate.

Create the backend cluster

Use the following command to create a backend cluster that connects to the Kafka servers you set up:

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:9092",
         "kafka2:9092",
         "kafka3:9092"
       ],
       "authentication": {
         "type": "anonymous"
       },
       "tls": {
         "enabled": false
       },
       "insecure_allow_anonymous_virtual_cluster_auth": true
     }' | jq -r ".id"
)

Create a virtual cluster

Create a virtual cluster that terminates oauth_bearer authentication and fetches the principal from the Kong Identity directory:

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": "oauth_bearer",
           "mediation": "terminate",
           "jwks": {
             "endpoint": "'$ISSUER_URL'/.well-known/jwks"
           },
           "fetch_kong_identity_principal": {
             "directory": "kong-identity-directory",
             "failure_mode": "error"
           }
         }
       ]
     }' | jq -r ".id"
)

For oauth_bearer authentication, the fetch_kong_identity_principal block doesn’t need a fetch_by field: the principal is always looked up by the JWT’s iss and sub claims.

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

This step requires a kafkactl version 5.17.0 or later. To check your version, run kafkactl version.

Note that this script is for demo purposes only and hard-codes the client ID, client secret, and scope. For production, we recommend securing sensitive data.

kafkactl generates tokens using a script. Create the script:

cat <<EOF > get-oauth-token.sh
#!/bin/bash
curl -s --fail -X POST "$ISSUER_URL/oauth/token" \
-H "Content-Type: application/x-www-form-urlencoded" \
-d "grant_type=client_credentials" \
-d "client_id=$CLIENT_ID" \
-d "client_secret=$CLIENT_SECRET" \
-d "scope=kafka" | jq -r '{"token": .access_token}'
EOF
chmod u+x get-oauth-token.sh

Create the kafkactl.yaml configuration:

cat <<EOF > kafkactl.yaml
contexts:
  direct:
    brokers:
      - localhost:9094
      - localhost:9095
      - localhost:9096
  vc:
    sasl:
      enabled: true
      mechanism: oauth
      tokenprovider:
        plugin: generic
        options:
          script: ./get-oauth-token.sh
          args: []
    brokers:
      - localhost:19092
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 validated the JWT against the auth server’s JWKS, looked up the principal in the kong-identity-directory Kong Identity directory by matching the token’s iss and sub against the oidc identity, attached the principal’s metadata to the connection, and applied the Modify Headers policy because context.auth.principal.metadata.team was operators.

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