Home / Event Gateway
Edit
Report

Encrypt Kafka message fields selected by schema registry tags with Event Gateway

Uses: Event Gateway Konnect API
Incompatible with
on-prem
Related Documentation
Event Gateway Documentation
Tags
#event-gateway #kafka #confluent #schema-registry #avro
Related Resources
Mask Kafka message fields selected by schema registry tags
Encrypt and decrypt Kafka fields in message values
Encrypt Fields policy
Decrypt Fields policy
Schema Registry entity
Static keys
Minimum Version
Event Gateway - 1.3
TL;DR

To encrypt Kafka message fields based on tags stored in a schema registry, you’ll need to:

  1. Register an Avro schema in a Confluent Schema Registry and tag the sensitive fields.
  2. Create a static key.
  3. Create a Schema Validation produce policy with a nested Encrypt Fields policy that computes the fields to encrypt from the schema tags.
  4. Create a Schema Validation consume policy with a nested Decrypt Fields policy that uses the same expression.

Prerequisites

Install kafkactl

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

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.1
    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.1
    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.1
    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.3.2
    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.

Overview

In this guide, you’ll learn how to encrypt fields of Kafka messages based on tags stored in a Confluent Schema Registry.

Instead of listing field paths in the policies, you tag the sensitive fields in the schema itself. Event Gateway reads the tags of each record’s schema, encrypts every tagged field before the record reaches Kafka, and decrypts the same fields for consumers. When the schema evolves and a new field gets tagged, Event Gateway encrypts it without any policy change.

We’ll use a customers topic that holds customer records with personal data. Producers and consumers that connect through the virtual cluster work with plaintext records, while the broker only stores the tagged fields in encrypted form. Anyone with direct access to the broker or to its backups can’t read the personal data.

Here’s how the data flows through the system:

 
sequenceDiagram
  autonumber
  participant producer as Producer
  participant egw as Event Gateway
  participant sr as Schema Registry
  participant broker as Kafka broker
  participant consumer as Consumer

  producer->>egw: produce record
  egw->>sr: get schema and PII tags
  egw->>egw: encrypt tagged fields
  egw->>broker: store record with encrypted fields
  consumer->>egw: consume request
  egw->>broker: fetch records
  broker->>egw: return records with encrypted fields
  egw->>sr: get schema and PII tags
  egw->>egw: decrypt tagged fields
  egw->>consumer: return plaintext record

The Encrypt Fields policy only works on string fields. A field that doesn’t exist is ignored. A field that isn’t a string is a policy failure, handled by the failure_mode setting.

Create a 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 clients connect to:

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": "customers_vc",
       "destination": {
         "id": "'$BACKEND_CLUSTER_ID'"
       },
       "dns_label": "customers",
       "authentication": [
         {
           "type": "anonymous"
         }
       ],
       "acl_mode": "passthrough"
     }' | jq -r ".id"
)

Create a listener with a forwarding policy

Create a listener to accept connections:

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": "customers_listener",
       "addresses": [
         "0.0.0.0"
       ],
       "ports": [
         "19092-19095"
       ]
     }' | jq -r ".id"
)

Create a Forward to Virtual Cluster policy to forward traffic 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_customers_vc",
       "config": {
         "type": "port_mapping",
         "advertised_host": "localhost",
         "destination": {
           "id": "'$VIRTUAL_CLUSTER_ID'"
         }
       }
     }'

Create a Schema Registry entity

Create a Schema Registry entity that points to the Confluent Schema Registry running locally. Since the Event Gateway data plane runs in the same Docker network as the Schema Registry, use the container hostname schema-registry:

curl -X POST "https://us.api.konghq.com/v1/event-gateways/$EVENT_GATEWAY_ID/schema-registries" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "name": "local-schema-registry",
       "type": "confluent",
       "config": {
         "schema_type": "avro",
         "endpoint": "http://schema-registry:8081",
         "timeout_seconds": 10
       }
     }'

Create a static key

Use OpenSSL to generate a 256-bit key that encrypts and decrypts the fields:

export MY_KEY="$(openssl rand -base64 32)"

Create a static key named my-key with the generated key:

curl -X POST "https://us.api.konghq.com/v1/event-gateways/$EVENT_GATEWAY_ID/static-keys" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "name": "my-key",
       "value": "'$MY_KEY'"
     }'

Register an Avro schema with PII tags

Register an Avro schema for the customers topic in the Confluent Schema Registry. The schema defines four fields: name, email, ssn, and city.

Alongside the schema, the registration body carries a metadata block that tags the three personal fields as PII:

cat <<EOF > schema.json
{
  "schemaType": "AVRO",
  "schema": "{\"type\": \"record\", \"name\": \"Customer\", \"fields\": [{\"name\": \"name\", \"type\": \"string\"}, {\"name\": \"email\", \"type\": \"string\"}, {\"name\": \"ssn\", \"type\": \"string\"}, {\"name\": \"city\", \"type\": \"string\"}]}",
  "metadata": {
    "properties": {"owner": "compliance-team"},
    "tags": {
      "name": ["PII"],
      "email": ["PII"],
      "ssn": ["PII"]
    }
  }
}
EOF
curl -sS --fail -X POST http://localhost:8081/subjects/customers-value/versions \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d @schema.json

In this configuration:

  • The subject name customers-value follows Confluent’s default TopicNameStrategy, which uses the pattern <topic>-value.
  • Each key of tags is a field path in the record, and each value lists the tags carried by that field. Use the same path that the field has in the message, for example profile.ssn for a field nested under profile.

Event Gateway passes the tag keys through as-is, so a key that doesn’t match a field path in the record is never encrypted.

Encrypt tagged fields on produce

Create a Schema Validation policy that parses Avro records against the Confluent Schema Registry during the produce phase. The Encrypt Fields policy needs a parsed record and the schema tags, so it must be nested under this policy:

PRODUCE_SCHEMA_VALIDATION_ID=$(curl -X POST "https://us.api.konghq.com/v1/event-gateways/$EVENT_GATEWAY_ID/virtual-clusters/$VIRTUAL_CLUSTER_ID/produce-policies" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "type": "schema_validation",
       "name": "validate_avro_produce",
       "config": {
         "type": "confluent_schema_registry",
         "schema_registry": {
           "name": "local-schema-registry"
         },
         "validate_value": true,
         "failure_mode": "reject"
       }
     }' | jq -r ".id"
)

The failure_mode: reject setting rejects a batch with a record that can’t be decoded or doesn’t conform to the schema. A record that Event Gateway can’t parse also can’t be encrypted, so it must not reach the broker.

Nest the Encrypt Fields policy under the Schema Validation policy. Instead of static paths, the paths field holds an expression that reads the PII tag from the record’s schema metadata and returns the tagged field paths:

curl -X POST "https://us.api.konghq.com/v1/event-gateways/$EVENT_GATEWAY_ID/virtual-clusters/$VIRTUAL_CLUSTER_ID/produce-policies" \
     --no-progress-meter --fail-with-body  \
     -H "Authorization: Bearer $KONNECT_TOKEN" \
     --json '{
       "type": "encrypt_fields",
       "name": "encrypt_pii_from_schema",
       "parent_policy_id": "'$PRODUCE_SCHEMA_VALIDATION_ID'",
       "config": {
         "failure_mode": "reject",
         "encrypt_fields": [
           {
             "paths": "record.value.schema.metadata.tags[\"PII\"]",
             "encryption_key": {
               "type": "static",
               "key": {
                 "name": "my-key"
               }
             }
           }
         ]
       }
     }'

In this configuration:

  • record.value.schema.metadata.tags is a map from tag name to the field paths carrying that tag. It’s populated by the parent Schema Validation policy, so the expression returns ["name", "email", "ssn"] for records that use the registered schema.
  • The encryption_key tells the policy to encrypt all returned paths with my-key. The city field from the schema isn’t tagged as PII, so it’s stored in plaintext.
  • failure_mode: reject means that any record the policy can’t encrypt never reaches the broker in plaintext.

Decrypt tagged fields on consume

Create a Schema Validation policy that parses Avro records during the consume phase:

CONSUME_SCHEMA_VALIDATION_ID=$(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": "schema_validation",
       "name": "validate_avro_consume",
       "config": {
         "type": "confluent_schema_registry",
         "schema_registry": {
           "name": "local-schema-registry"
         },
         "validate_value": true,
         "failure_mode": "skip"
       }
     }' | jq -r ".id"
)

Create the Decrypt Fields policy nested under the Schema Validation policy. It uses the same expression, so it decrypts the same fields that the Encrypt Fields policy encrypted:

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": "decrypt_fields",
       "name": "decrypt_pii_from_schema",
       "parent_policy_id": "'$CONSUME_SCHEMA_VALIDATION_ID'",
       "config": {
         "failure_mode": "mark",
         "key_sources": [
           {
             "type": "static"
           }
         ],
         "decrypt_fields": {
           "paths": "record.value.schema.metadata.tags[\"PII\"]"
         }
       }
     }'

The Decrypt Fields policy reads the key reference from the kong/enc header that the Encrypt Fields policy adds to each record, and looks up the key in the static keys.

The policy uses failure_mode: mark. A failed decryption never exposes plaintext, so the record is delivered with the fields still encrypted and a kong/policy-failure-<id> header that explains the failure.

Configure kafkactl

Create a kafkactl configuration with a direct context that connects to Kafka, and a vc context that connects through the virtual cluster. Both contexts point to the Schema Registry, so kafkactl can serialize records to Avro and deserialize them for display:

cat <<EOF > kafkactl.yaml
contexts:
  direct:
    brokers:
      - localhost:9094
      - localhost:9095
      - localhost:9096
    schemaRegistry:
      url: http://localhost:8081
  vc:
    brokers:
      - localhost:19092
    schemaRegistry:
      url: http://localhost:8081
EOF

Create a topic and produce records

Create the customers topic:

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

Produce two customer records through the virtual cluster. kafkactl serializes them to Avro using the schema from the registry:

echo '{"name":"John Doe","email":"john.doe@example.com","ssn":"098-76-5432","city":"San Francisco"}
{"name":"Maria Silva","email":"maria.silva@example.com","ssn":"123-45-6789","city":"Boston"}' | kafkactl -C kafkactl.yaml --context vc produce customers

Validate

Consume directly from Kafka

Consume the records straight from the broker, with their headers, to confirm that Kafka stores the tagged fields encrypted:

kafkactl -C kafkactl.yaml --context direct consume customers --from-beginning --exit --print-headers

The name, email, and ssn fields are encrypted, and city is in plaintext. Each record carries a kong/enc header that identifies the static key by its ID:

kong/enc:  -static://<static-key-id>#{"name":"AOrXosB8U5J2\/pSY4WSJFCe\/DJCW5noJ0F+H1fg\/CY6y3F2WxA==","email":"ABIVUSkoQFyoECpFYkf6zYPUZOgW7Dw\/Nee8XKeDNHuODq7DjLAou4cXtEb4tRCANw==","ssn":"ALVdciwDnNQHwH3YdEHmDGGqDhkNwXfztmWjsOT34Ql0OdxzXkhq0w==","city":"San Francisco"}
kong/enc:  -static://<static-key-id>#{"name":"AFvtXJdxtNR+GNIJ8DHCsd6mYQksB3maNBiPsz2zrxoTBoDh\/0NsSw==","email":"AJTOt1CUPeATneZNxUMKCqxIUS0S+zryDXFs6G\/lYv65zmLeLCXqDmq2dI8MDh5WYWxbGQ==","ssn":"AOIAev91oOpQ9YVotwe8bm5ZHVVpfTIhTwjDeoEjjkJJdpuchQ6f2g==","city":"Boston"}

Consume through the virtual cluster

Consume the same records through the virtual cluster:

kafkactl -C kafkactl.yaml --context vc consume customers --from-beginning --exit

The tagged fields are decrypted, and consumers receive the original records:

{"name":"John Doe","email":"john.doe@example.com","ssn":"098-76-5432","city":"San Francisco"}
{"name":"Maria Silva","email":"maria.silva@example.com","ssn":"123-45-6789","city":"Boston"}

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