Data security Interceptors

Data security Interceptor reference for Conduktor Gateway: field-level encryption, full payload encryption, data masking.

Configure audit logging, data masking, header injection, the message integrity Interceptor and encryption Interceptors to secure and track data in your Kafka topics.

Job to do	Interceptor to use
Track the APIs used by an application	Audit
Hide sensitive fields from consumers	Data masking
Add metadata to messages	Header injection
Sign and verify message integrity	Message integrity Interceptor
Encrypt data	Encryption
Keep data usable for analytics while protecting it	Tokenization
Delete a single customer’s data from Kafka	Crypto shredding

From our blog: FedRAMP High for Kafka without replatforming Meet FedRAMP High encryption requirements on your existing Kafka, no replatforming.

Kafka and HIPAA 2026 What the 2026 HIPAA Security Rule changes for Kafka, and how to close the gaps.

Common configuration

The following configurations are shared across multiple Interceptors:

Environment variables as secrets

To ensure your secrets don’t appear in your Interceptors, you can refer to the environment variables set in your Gateway container.

Use the format ${MY_ENV_VAR}.

We recommend using this for schema registry or Vault secrets and any other values you’d like to hide in the configuration.

Schema registry

Gateway reads schemas either through an external connection or from configuration on the Interceptor itself.

Schema-encoded records whose schema ID is stored in a record header, the format introduced in Confluent Platform 8.2, require Gateway 3.21.0 or later.

External connection

External connections are a preview feature and subject to change. Inline schemaRegistryConfig remains fully supported.

Set schemaRegistryConnectionName to the name of a schema registry external connection, declared once when Gateway starts with variables that follow this pattern:

GATEWAY_EXTERNAL_CONNECTION_<NAME>_CONFIG_<PROPERTY>

Key	Type	Description
schemaRegistryConnectionName	string	The <NAME> of a schema registry connection. Gateway rejects an unknown name with 400 Bad Request.

For a connection declared as GATEWAY_EXTERNAL_CONNECTION_DEFAULT_SR_CONFIG_*, the Interceptor reaches it with:

"schemaRegistryConnectionName": "DEFAULT_SR"

Set either schemaRegistryConnectionName or schemaRegistryConfig, never both. Gateway rejects a configuration that sets both with 400 Bad Request, rather than silently picking one.

Remove a connection an Interceptor already names and the Interceptor stays visible through the API but stops applying to traffic. See what happens at startup.

Inline configuration

Gateway supports Confluent-like and AWS Glue schema registries.

Key	Type	Default	Description
type	string	CONFLUENT	The type of schema registry to use: choose CONFLUENT (for Confluent-like schema registries including OSS Kafka) or AWS for AWS Glue schema registries.
additionalConfigs	map		Additional properties maps to specific security-related parameters. For enhanced security, you can hide the sensitive values using environment variables as secrets.
Confluent-like			Configuration for Confluent-like schema registries
host	string		URL of your schema registry.
cacheSize	string	50	Number of schemas that can be cached locally by this Interceptor so that it doesn’t have to query the schema registry every time.
AWS Glue			Configuration for AWS Glue schema registries
region	string		The AWS region for the schema registry, e.g. us-east-1.
registryName	string		The name of the schema registry in AWS (leave blank for the AWS default of default-registry).
basicCredentials	string		Access credentials for AWS.
AWS credentials			AWS credential configuration
accessKey	string		The access key for the connection to the schema registry.
secretKey	string		The secret key for the connection to the schema registry.
validateCredentials	bool	true	Whether Gateway rejects a blank accessKey or secretKey when it builds the credentials. Set it to false to accept them and fail later instead.
accountId	string		The Id for the AWS account to use.

If you don’t supply a basicCredentials section for the AWS Glue schema registry, the client will attempt to find the connection information it needs from the environment (see AWS docs for details ) and the credentials required can be passed this way to Gateway as part of its core configuration.

Read our blog about schema registry.

Audit Interceptor

This Interceptor logs information from API key requests. To use it, inject it and implement ApiKeyAuditLog interface for audit.

The currently supported Kafka API requests are:

• ProduceRequest (PRODUCE)
• FetchRequest (FETCH)
• CreateTopicRequest (CREATE_TOPICS)
• DeleteTopicRequest (DELETE_TOPICS)
• AlterConfigRequest (ALTER_CONFIGS)

When incremental fetch sessions are enabled, Gateway records a fetch only when the request lists the topics being read, which a consumer stops doing once it settles into steady-state reads. A long-lived consumer can appear idle in audit logs while it keeps consuming. This affects audit visibility only — it doesn’t change the data a consumer receives.

Configure audit Interceptor

Name	Type	Default	Description
topic	String	.*	Topics that match this regex will have the Interceptor applied
apiKeys	Set[string]		Set of Kafka API keys to be audited
vcluster	String	.*	vcluster that matches this regex will have the Interceptor applied
username	String	.*	username that matches this regex will have the Interceptor applied
consumerGroupId	String	.*	consumerGroupId that matches this regex will have the Interceptor applied
topicPartitions	Set[Integer]		Set of topic partitions to be audited

Audit Interceptor example

curl

curl \
  --request PUT \
  --url 'http://localhost:8888/gateway/v2/interceptor' \
  --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
  --header 'Content-Type: application/json' \
  --data-raw '{
  "kind": "Interceptor",
  "apiVersion": "gateway/v2",
  "metadata": {
    "name": "myAuditInterceptorPlugin"
  },
  "spec": {
    "pluginClass": "io.conduktor.gateway.interceptor.AuditPlugin",
    "priority": 100,
    "config": {
      "topic": ".*",
      "apiKeys": [
        "PRODUCE",
        "FETCH"
      ],
      "vcluster": ".*",
      "username": ".*",
      "consumerGroupId": ".*",
      "topicPartitions": [
        1,
        2
      ]
    }
  }
}'

Conduktor CLI

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: myAuditInterceptorPlugin
  scope:
    vCluster: passthrough
spec:
  pluginClass: io.conduktor.gateway.interceptor.AuditPlugin
  priority: 100
  config:
    topic: ".*"
    apiKeys:
      - PRODUCE
      - FETCH
    vcluster: ".*"
    username: ".*"
    consumerGroupId: ".*"
    topicPartitions:
      - 1
      - 2

Apply with:

conduktor apply -f audit-interceptor.yaml

Data masking Interceptor

Field level data masking Interceptor masks sensitive fields within messages as they are consumed.

Configure data masking Interceptor

The policies will be applied when consuming messages.

Key	Type	Default	Description
topic	String	.*	Topics that match this regex will have the Interceptor applied.
policies	Policy list		List of your masking policies.
errorPolicy	String	fail_fetch	Determines the plugin behavior when it can’t parse a fetched message without an associated schema: fail_fetch or skip_masking.
schemaRegistryConfig	Schema registry		The schema registry in use. Required for Avro, JSON Schema or Protobuf data.
schemaRegistryConnectionName	String		Name of a schema registry external connection. An alternative to schemaRegistryConfig — set one or the other, not both.

Data masking policy

Key	Type	Description
name	String	Unique name to identify your policy.
fields	Set of String	Set of fields that should be obfuscated with the masking rule. Fields can be in a nested structure with dot .. For example: education.account.username, banks[0].accountNo or banks[*].accountNo.
rule	Rule	Masking rule to apply.

Data masking rule

Key	Type	Default	Description
type	Masking type	MASK_ALL	The type of masking (see below).
maskingChar	String	*	The character(s) used for masking data.
numberOfChars	number		Number of masked characters, required if type != MASK_ALL.

Masking type

• MASK_ALL: all data will be masked
• MASK_FIRST_N: the first n characters will be masked
• MASK_LAST_N: the last n characters will be masked

Error policy

You can control the plugin behavior when it can’t parse a fetched message through its errorPolicy which can be set to fail_fetch or skip_masking.

The error policy only applies to messages that do not have an associated schema. When a message has a schema (Avro, JSON Schema or Protobuf), the plugin uses the schema to parse the message and the error policy is not triggered.

The default is fail_fetch. In this mode, the plugin will return a failure to read the batch which the fetch record is part of, effectively blocking any consumer.

In skip_masking mode, if there’s a failure to parse a message being fetched (e.g. an encrypted record or a schemaless message that can’t be parsed), then that record is skipped and returned un-masked.

Full payload encryption compatibility

Data masking is compatible with full payload encryption. When both Interceptors are applied to the same topic, data masking automatically detects records with full payload encryption headers and skips them, preventing deserialization errors that would otherwise occur when attempting to mask encrypted content.

Check out the encryption configuration for details.

Field level encryption is not affected by this behavior.

Data masking Interceptor example

curl

curl \
  --request PUT \
  --url 'http://localhost:8888/gateway/v2/interceptor' \
  --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
  --header 'Content-Type: application/json' \
  --data-raw '{
  "kind": "Interceptor",
  "apiVersion": "gateway/v2",
  "metadata": {
    "name": "myFieldLevelDataMaskingPlugin"
  },
  "spec": {
    "pluginClass": "io.conduktor.gateway.interceptor.FieldLevelDataMaskingPlugin",
    "priority": 100,
    "config": {
      "schemaRegistryConfig": {
        "host": "http://schema-registry:8081"
      },
      "policies": [
        {
          "name": "Mask password",
          "rule": {
            "type": "MASK_ALL"
          },
          "fields": [
            "password"
          ]
        },
        {
          "name": "Mask visa",
          "rule": {
            "type": "MASK_LAST_N",
            "maskingChar": "X",
            "numberOfChars": 4
          },
          "fields": [
            "visa"
          ]
        }
      ]
    }
  }
}'

Conduktor CLI

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: myFieldLevelDataMaskingPlugin
  scope:
    vCluster: passthrough
spec:
  pluginClass: io.conduktor.gateway.interceptor.FieldLevelDataMaskingPlugin
  priority: 100
  config:
    schemaRegistryConfig:
      host: http://schema-registry:8081
    policies:
      - name: Mask password
        rule:
          type: MASK_ALL
        fields:
          - password
      - name: Mask visa
        rule:
          type: MASK_LAST_N
          maskingChar: X
          numberOfChars: 4
        fields:
          - visa

Apply with:

conduktor apply -f data-masking-interceptor.yaml

Secured schema registry

curl

curl \
  --request PUT \
  --url 'http://localhost:8888/gateway/v2/interceptor' \
  --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
  --header 'Content-Type: application/json' \
  --data-raw '{
  "kind": "Interceptor",
  "apiVersion": "gateway/v2",
  "metadata": {
    "name": "myFieldLevelDataMaskingPlugin"
  },
  "spec": {
    "pluginClass": "io.conduktor.gateway.interceptor.FieldLevelDataMaskingPlugin",
    "priority": 100,
    "config": {
      "schemaRegistryConfig": {
        "host": "http://schema-registry:8081",
        "additionalConfigs": {
          "schema.registry.url": "${SR_URL}",
          "basic.auth.credentials.source": "${SR_BASIC_AUTH_CRED_SRC}",
          "basic.auth.user.info": "${SR_BASIC_AUTH_USER_INFO}"
        }
      },
      "policies": [
        {
          "name": "Mask password",
          "rule": {
            "type": "MASK_ALL"
          },
          "fields": [
            "password"
          ]
        },
        {
          "name": "Mask visa",
          "rule": {
            "type": "MASK_LAST_N",
            "maskingChar": "X",
            "numberOfChars": 4
          },
          "fields": [
            "visa"
          ]
        }
      ]
    }
  }
}'

Conduktor CLI

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: myFieldLevelDataMaskingPlugin
  scope:
    vCluster: passthrough
spec:
  pluginClass: io.conduktor.gateway.interceptor.FieldLevelDataMaskingPlugin
  priority: 100
  config:
    schemaRegistryConfig:
      host: http://schema-registry:8081
      additionalConfigs:
        schema.registry.url: $${SR_URL}
        basic.auth.credentials.source: $${SR_BASIC_AUTH_CRED_SRC}
        basic.auth.user.info: $${SR_BASIC_AUTH_USER_INFO}
    policies:
      - name: Mask password
        rule:
          type: MASK_ALL
        fields:
          - password
      - name: Mask visa
        rule:
          type: MASK_LAST_N
          maskingChar: X
          numberOfChars: 4
        fields:
          - visa

Apply with:

conduktor apply -f data-masking-secured-sr-interceptor.yaml

Dynamic header injection Interceptor

This Interceptor injects headers (such as user IP) into records as they pass through Gateway. Use target to choose whether Gateway injects them when clients produce, when they consume, or both. We support templating in this format: X-CLIENT_IP: "{{userIp}} testing".

Choose the produce or consume path

The target setting controls which path the Interceptor runs on.

target	What Gateway does	Use it when
PRODUCE (default)	Injects headers when clients produce through Gateway	producers go through Gateway and you want the header stored in the topic
CONSUME	Injects headers into the fetch response when clients consume	producers bypass Gateway, so consume is the only path Gateway sees
BOTH	Injects on both paths	some producers go through Gateway and some don’t, and you want every record tagged when it’s read

In a future version we will stop defaulting target to PRODUCE and so we log a warning when it is not set. We advise you to set it explicitly.

On consume, Gateway adds the headers to the fetch response but does not write them to the topic, so the record in Kafka stays unchanged.

With BOTH, Gateway skips a header on consume if the record already has it, so records aren’t tagged twice. Set overrideIfExists to true to inject anyway.

Context variables

These values are available as template variables:

• uuid
• userIp
• vcluster
• user
• clientId
• gatewayIp
• gatewayHost
• gatewayVersion
• apiKey
• apiKeyVersion
• timestampMillis

Context variables resolve relative to the path the Interceptor runs on. With target: CONSUME or BOTH, clientId, user, userIp and vcluster describe the consumer rather than the producer, and apiKey is Fetch instead of Produce. Gateway regenerates uuid and timestampMillis on every fetch, so each read of the same record gets different values.

When Gateway sits behind a load balancer, userIp resolves to the original client address only when Gateway captures it. Otherwise, it’s the load balancer’s address.

Record extraction templates

You can also extract fields from the record key or value:

• {{record.key}} - extract the entire key payload as a string
• {{record.value}} - extract the entire value payload as a string
• {{record.key.fieldName}} - extract a specific field from the record key
• {{record.value.fieldName}} - extract a specific field from the record value

For example, if your record has a key with a field named “id”, you can use {{record.key.id}} to extract that value and inject it as a header.

To use field extraction (record.key.fieldName or record.value.fieldName) with Avro, JSON Schema or Protobuf data, give the Interceptor a schema registry so Gateway can deserialize the records — either schemaRegistryConnectionName or schemaRegistryConfig. For plain JSON data, no schema registry is needed.

Configure header injection Interceptor

Config	Type	Default	Description
topic	String		Regular expression that matches the topics the Interceptor applies to.
target	Enum	PRODUCE	Which path to inject on: PRODUCE, CONSUME or BOTH. See Choose the produce or consume path.
headers	Map		Map of header key and header value to inject. Values can use template variables like {{userIp}} or record extraction patterns.
overrideIfExists	boolean	false	Whether to override headers that already exist on the record.
failOnError	boolean	false	Whether to reject the request when header injection fails. When false, Gateway logs the error and continues. See Error handling.
schemaRegistryConfig	Schema registry		Schema Registry configuration. Required when using record.key.field or record.value.field templates with schema-encoded data.
schemaRegistryConnectionName	String		Name of a schema registry external connection. An alternative to schemaRegistryConfig — set one or the other, not both.

Error handling

The failOnError setting controls how the Interceptor handles errors during header injection, such as a missing field or a payload it can’t deserialize.

failOnError	On produce	On consume
false (default)	Gateway logs a warning, skips that header and continues	Same as produce
true	The request fails and the producer gets an error	Gateway fails the whole fetch response. The consumer gets no records and no error, so it stalls while Gateway logs an error on each retry.

If target is CONSUME or BOTH, we recommend leaving failOnError at false. To fail produce requests but only log on consume, use two Interceptors with different names: one with target: PRODUCE and failOnError: true, and one with target: CONSUME and failOnError: false.

Header injection Interceptor example

curl

curl \
  --request PUT \
  --url 'http://localhost:8888/gateway/v2/interceptor' \
  --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
  --header 'Content-Type: application/json' \
  --data-raw '{
  "kind": "Interceptor",
  "apiVersion": "gateway/v2",
  "metadata": {
    "name": "myDynamicHeaderInjectionInterceptor"
  },
  "spec": {
    "pluginClass": "io.conduktor.gateway.interceptor.DynamicHeaderInjectionPlugin",
    "priority": 100,
    "config": {
      "topic": "topic.*",
      "target": "PRODUCE",
      "headers": {
        "X-CLIENT_IP": "{{userIp}} testing",
        "X-USER-ID": "{{record.key.id}}",
        "X-USER-EMAIL": "{{record.value.email}}"
      },
      "overrideIfExists": true,
      "failOnError": false,
      "schemaRegistryConfig": {
        "host": "http://schema-registry:8081"
      }
    }
  }
}'

Conduktor CLI

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: myDynamicHeaderInjectionInterceptor
  scope:
    vCluster: passthrough
spec:
  pluginClass: io.conduktor.gateway.interceptor.DynamicHeaderInjectionPlugin
  priority: 100
  config:
    topic: topic.*
    target: PRODUCE
    headers:
      X-CLIENT_IP: "{{userIp}} testing"
      X-USER-ID: "{{record.key.id}}"
      X-USER-EMAIL: "{{record.value.email}}"
    overrideIfExists: true
    failOnError: false
    schemaRegistryConfig:
      host: http://schema-registry:8081

Apply with:

conduktor apply -f header-injection-interceptor.yaml

Let’s produce a simple record to the injectHeaderTopic topic.

echo 'inject_header' | docker-compose exec -T kafka-client \
    kafka-console-producer  \
        --bootstrap-server conduktor-gateway:6969 \
        --producer.config /clientConfig/gateway.properties \
        --topic injectHeaderTopic

Let’s consume from our injectHeaderTopic.

docker-compose exec kafka-client \
  kafka-console-consumer \
    --bootstrap-server conduktor-gateway:6969 \
    --consumer.config /clientConfig/gateway.properties \
    --topic injectHeaderTopic \
    --from-beginning \
    --max-messages 1 \
    --property print.headers=true

You should see the message with headers as below

X-USER_IP:172.19.0.3 testing   inject_header

Message integrity Interceptor

The message integrity Interceptor signs Kafka records on produce and verifies them on fetch, letting consumers detect whether a record changed after it was produced. Two plugins work together:

• ProduceIntegrityPolicyPlugin signs records using HMAC-SHA256 through Google Tink.
• FetchIntegrityPolicyPlugin verifies signatures and drops or allows records based on your policy.

You store signing keys in your HashiCorp Vault Key-Value (KV) v2 instance. Gateway reads and caches them locally.

Ordering with other Interceptors

The message integrity Interceptor is always the outermost layer — signing runs last on produce (after all other Interceptors have transformed the record) and verification runs first on fetch (before any normal Interceptor runs). This ensures the signature covers the final produced payload and is verified before any transformation on consume.

Gateway enforces this ordering automatically. You don’t have to set specific priority values for integrity Interceptors — Gateway places them in fixed pipeline positions regardless of their configured priority:

• Produce: the sign plugin always runs after all other Interceptors.
• Fetch: the verify plugin always runs before all other Interceptors.

Gateway also validates that two integrity Interceptors of the same type don’t have overlapping scopes (Virtual Cluster, group or username). If they do, Gateway rejects the configuration.

The Interceptor reads Vault credentials only from its config. You have to specify credential fields (such as token, roleId or secretId) in the config. You can set their values with placeholders like token: "${VAULT_TOKEN}", which Gateway resolves when it loads the config. Gateway does not fall back to environment variable names when you omit a credential field (for example, there is no built-in fallback to VAULT_TOKEN if you leave out token).

When Vault is unreachable (for example, on a cache miss), Gateway propagates the error to the Kafka client: producers receive the failure on produce and consumers receive it on fetch. Gateway does not silently drop records in these cases.

Configure the secretKeyUri

secretKeyUri points to a specific field in a KV v2 secret. Use the format <mount>/data/<path>#<fieldName>.

Examples: secret/data/signing-key#key, secret/data/app/keys/signing#hmacKey.

Manage key versions in Vault KV v2

Vault KV v2 versions secrets: each write to the same path creates a new version. Gateway handles versions as follows:

• Produce (sign): Gateway uses the latest version of the secret at the path you configured. When you write a new value to that path in Vault, Gateway picks it up once the cache entry expires (see cache.ttlMs). Each signed record stores the key version in its signature header.
• Fetch (verify): The signature on each record identifies the key version used to sign it. Gateway fetches that exact version from Vault to verify, so records signed with an older version still verify correctly after you rotate to a newer version.
• Older versions: Keep older secret versions readable in Vault until you no longer need to verify records signed with them (for example, until they are consumed or past your retention period).

Configure produce (sign) plugin

If a record already has the signature header, the Interceptor throws PolicyViolationException and does not re-sign.

Key	Type	Default	Description
topic	String	.*	Topics matching this regex have the Interceptor applied
signatureHeader	String	conduktor.integrity.signature	Header name for the signature
secretKeyUri	String	—	Vault KV v2 path and field (for example, secret/data/signing-key#key)
keyProviderConfig	Object	—	Key provider (Vault); see Authenticate with Vault below
cache.ttlMs	Long	300000	How long (in milliseconds) the signing key is cached before Gateway fetches it again from Vault. A shorter TTL means rotated keys are picked up faster (default: five minutes)
cache.maxSize	Integer	100	Maximum number of keys to cache

Configure fetch (verify) plugin

Gateway drops records that fail verification (missing signature, malformed header or invalid MAC (Message Authentication Code)) and emits an audit event. After successful verification, Gateway removes the signature header from the record before returning it to the consumer. For other errors (such as Vault being unreachable on a cache miss), Gateway propagates the error to the Kafka client and does not silently drop the record.

When a record is dropped or allowed with missing signature, the fetch (verify) plugin emits a fetch response audit event (error level). To receive these events, enable the audit feature with GATEWAY_FEATURE_FLAGS_AUDIT (see Audit logs and Environment variables).

Audit event details:

• Event type: fetch response audit event (level: error)
• Information included: topic, partition, offset, Interceptor name, plugin name (FetchIntegrityPolicyPlugin) and a message describing the reason
• Reason values: missing_signature (no signature header), malformed_signature (header could not be decoded), verification_failed:INVALID_SIGNATURE (MAC does not match) or verification_failed:UNKNOWN_KEY (key version not found in Vault)

Key	Type	Default	Description
topic	String	.*	Topics matching this regex have the Interceptor applied
signatureHeader	String	conduktor.integrity.signature	Header name where Gateway stores the signature
missingSignaturePolicy	String	SKIP	When a record has no signature header: SKIP (drop and audit) or ALLOW (audit and keep in response)
keyProviderConfig	Object	—	Key provider (Vault); see Authenticate with Vault below
cache.ttlMs	Long	300000	How long (in milliseconds) a verification key stays in cache before Gateway fetches it again from Vault. This controls how long previously seen keys are reused without a Vault lookup
cache.maxSize	Integer	100	Maximum number of keys to cache

Authenticate with Vault for message integrity

All auth types use the common fields: uri (required) and optionally namespace, openTimeoutSeconds (default five), readTimeoutSeconds (default 30), keyStore, trustStore and connectionBackoff. Set type to one of the following and add the corresponding fields.

You can also configure TLS for the Vault connection:

• keyStore: set keyStorePath and keyStorePassword for client certificate authentication
• trustStore: set trustStorePath and trustStorePassword to verify the Vault server certificate

type	Required fields	Optional fields
TOKEN	token	—
APP_ROLE	roleId, secretId	mount (default approle)
KUBERNETES	role	tokenPath (default /var/run/secrets/kubernetes.io/serviceaccount/token), mount (default kubernetes)
USERNAME_PASSWORD	username, password	mount (default userpass)
GITHUB	token	mount (default github)
LDAP	username, password	mount (default ldap)
GCP	role, jwt	—
AWS_EC2_PKCS7	pkcs7	role, nonce, mount (default aws)
AWS_EC2	identity, signature	role, nonce, mount (default aws)
AWS_IAM	iamRequestUrl, iamRequestBody, iamRequestHeaders	role, mount (default aws)
JWT	provider, role, jwt	—

All auth types except TOKEN support automatic token renewal. Gateway renews Vault tokens in the background so your Interceptor continues to work without interruption.

Optional connectionBackoff (for transient Vault failures): backoffDelay (default five), backoffMaxDelay (default 30), backoffChronoUnit (default SECONDS), backoffDelayFactor (default 1.1).

Set up Vault for message integrity

1. Enable KV v2: vault secrets enable -version=2 kv
2. Create a signing key for HMAC-SHA256:
   • The key material has to be at least 32 bytes (256 bits) after decoding. Gateway enforces this per NIST SP 800-107 Rev 1 and rejects shorter keys with an error.
   • Store the key in a KV v2 secret as a base64-encoded string. Gateway decodes the Base64 value and uses the resulting bytes, so the decoded length has to be at least 32 bytes.
   • The field name in the secret has to match the <fieldName> in your secretKeyUri (for example, secret/data/signing-key#key uses field name key).
   • Example: generate 32 random bytes, base64-encode them for storage, then write to Vault: KEY=$(openssl rand -base64 32) then vault kv put -mount=secret signing-key key="$KEY"
3. Create a policy for Gateway with read access on the secret path (for example, path "secret/data/signing-key" { capabilities = ["read"] })

Understand the signature format

Gateway stores each signature in a Kafka header as JSON with two fields: k = the secretKeyUri with its version (for example, secret/data/signing-key#key@1) and s = the base64-encoded HMAC-SHA256 MAC.

Message integrity Interceptor examples

Produce (sign) – Conduktor CLI

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: integrity-sign
  scope:
    vCluster: passthrough
spec:
  pluginClass: io.conduktor.gateway.interceptor.integrity.ProduceIntegrityPolicyPlugin
  config:
    topic: ".*"
    signatureHeader: "conduktor.integrity.signature"
    secretKeyUri: "secret/data/signing-key#key"
    keyProviderConfig:
      vault:
        uri: "http://localhost:8200"
        type: TOKEN
        token: "${VAULT_TOKEN}"
    cache:
      ttlMs: 3600000
      maxSize: 100

Fetch (verify) – Conduktor CLI

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: integrity-verify
  scope:
    vCluster: passthrough
spec:
  pluginClass: io.conduktor.gateway.interceptor.integrity.FetchIntegrityPolicyPlugin
  config:
    topic: ".*"
    signatureHeader: "conduktor.integrity.signature"
    missingSignaturePolicy: SKIP
    keyProviderConfig:
      vault:
        uri: "http://localhost:8200"
        type: TOKEN
        token: "${VAULT_TOKEN}"
    cache:
      ttlMs: 3600000
      maxSize: 100

Apply with: conduktor apply -f integrity-sign-interceptor.yaml and conduktor apply -f integrity-verify-interceptor.yaml.

Encryption Interceptors

Gateway encrypts your Kafka data as it passes through the proxy, before it reaches the broker.

Unlike TLS (Transport-Level Encryption), Gateway encryption ensures data remains encrypted when stored on Kafka brokers.

The section covers all of the configuration options available for every encryption Interceptor.

You can also check out other resources:

• encryption use cases
• encryption examples

Encryption configuration

The properties detailed in this section work for the following plugins:

	Plugin
List-based	EncryptPlugin
Schema-based	EncryptSchemaBasedPlugin

Both schema-based and list-based encryption plugins have their configuration, but some properties are common to both of them.

Key	Type	Default	Description
Common properties
topic	String	.*	Topics matching this regex will have the interceptor applied.
schemaRegistryConfig	SchemaRegistry		Configuration of your schema registry. Required if you want to encrypt data produced using Avro, JSON or Protobuf schemas.
schemaRegistryConnectionName	String		Name of a schema registry external connection. An alternative to schemaRegistryConfig — set one or the other, not both.
schemaDataMode	String	preserve_avro	Controls the format Gateway writes when it encrypts an Avro record. preserve_avro keeps the record in Avro, so it stays linked to its schema in the backing topic. This is the default and the recommended value since Gateway v3.3. convert_json is deprecated and will be removed in Gateway 3.24.0. It converts the record to JSON and breaks the link to its schema in the backing topic. Existing configurations continue to work, and Gateway logs a deprecation warning when it loads them.
kmsConfig	KMS		Configuration of one or multiple KMS.
kmsConnectionName	String		Name of a KMS external connection. An alternative to kmsConfig — set one or the other, not both
enableAuditLogOnError	Boolean	true	The audit log will be enabled when an error occurs during encryption/decryption
compressionType	Enum	none	The data is compressed before encryption (only for data configured with full payload encryption). Available values are: none, gzip, snappy, lz4 or zstd.
throttleTimeMs	Integer	0	When encryption fails, apply client throttling for the specified time in milliseconds. This helps prevent overwhelming the system during encryption failures. Find out more about client throttling
errorPolicy	String	fail_on_encrypted	Determines the plugin behavior when it encounters a record that’s already encrypted. Possible values: fail_on_encrypted, skip_already_encrypted
List-based properties
recordValue	Value and key encryption		Configuration to encrypt the record value.
recordKey	Value and key encryption		Configuration to encrypt the record key.
recordHeader	Headers encryption		Configuration to encrypt the record headers.
Schema-based properties
defaultKeySecretId	Key reference		Default key URI to use if none is set in the schema, when the Interceptor configures its KMS inline. Can be a template for Crypto Shredding use cases.
defaultKeyId	Key reference		Bare key name for every encrypted field, when the Interceptor names a KMS connection.
defaultAlgorithm	Algorithm	AES128_GCM	Default algorithm to use if no algorithm is set in the schema.
tags	List[String]		List of tags to search for in the schema to encrypt the specified fields.
namespace	String	conduktor.	Prefix of custom schema constraints for encryption.

List-based

Decide what you want to encrypt:

• Record value and record key:
  • Encrypt a set of fields
  • Encrypt the full payload
• or header keys:
  • Encrypt a set of fields
  • Encrypt the full payload
  • Encrypt a set of headers that match a regex

Record values and record keys

Set the following properties for recordValue (value encryption) and/or recordKey (key encryption):

Key	Type	Default	Description
Full-payload encryption
payload.keySecretId	Key reference		Key URI, when the Interceptor configures its KMS inline. Can be a template for Crypto Shredding use cases.
payload.keyId	Key reference		Bare key name, when the Interceptor names a KMS connection. Set this or keySecretId, never both.
payload.algorithm	Algorithm	AES128_GCM	Algorithm to leverage.
Field-Level encryption
fields[].fieldName	String		Name of the field to encrypt. It can be a nested structure with a dot . such as education.account.username or banks[0].accountNo.
fields[].keySecretId	Key reference		Key URI, when the Interceptor configures its KMS inline. Can be a template for crypto shredding use cases.
fields[].keyId	Key reference		Bare key name, when the Interceptor names a KMS connection. Set this or keySecretId, never both.
fields[].algorithm	Algorithm		Algorithm to use to encrypt this field.

Check out the encryption examples.

Header keys

Set the following properties for recordHeader:

Key	Type	Default	Description
Full-payload encryption			Configuration to encrypt the full payload.
payload.keySecretId	Key reference		Key URI, when the Interceptor configures its KMS inline. Can be a template for crypto shredding use cases.
payload.keyId	Key reference		Bare key name, when the Interceptor names a KMS connection. Set this or keySecretId, never both.
payload.algorithm	Algorithm	AES128_GCM	Algorithm to leverage.
Field-level encryption
fields[].fieldName	String		Name of the field to encrypt. It can be a nested structure with a dot . such as education.account.username or banks[0].accountNo.
fields[].keySecretId	Key reference		Key URI, when the Interceptor configures its KMS inline. Can be a template for crypto shredding use cases.
fields[].keyId	Key reference		Bare key name, when the Interceptor names a KMS connection. Set this or keySecretId, never both.
fields[].algorithm	Algorithm
Headers encryption
header	String		Headers that match this regex will be encrypted. It can encrypt all headers including Gateway headers.

Check out the encryption example.

Schema-based

In order to encrypt your data, you can set a few constraints in your schema. These constraints are detailed below, assuming you’re using the default namespace value which is conduktor.​. If you have changed the namespace value in the Interceptor configuration, please change the key name in your schema accordingly.

Key	Type	Default	Description
conduktor.keySecretId	Key reference		Key URI, when the Interceptor configures its KMS inline. Can be a template for crypto shredding use cases.
conduktor.algorithm	Algorithm	AES128_GCM	Algorithm to use to encrypt this field.
conduktor.tags	List[String]		Fields tagged with a matching tag from your Interceptor will be encrypted using the key and algorithm specified in the schema. If these are not defined in the schema, the defaultKeySecretId (or defaultKeyId) and defaultAlgorithm from the Interceptor configuration will be used.

If your field meets one of these three conditions, then it will be encrypted:

1. This field has a keySecretId set in the schema
2. This field has a algorithm set in the schema
3. This field has a set of tags set in the schema, and one of them is part of the tags list specified in the Interceptors.

Configure the KMS inline for schema-based encryption. A KMS external connection doesn’t yet honor per-field keys from a schema.

Check out the encryption example.

Secret keys

Mustache template

In all the encryption plugins, you can use mustache templates for the key an entry names — keySecretId inline, or keyId when the Interceptor names a KMS external connection. That way, your keys are dynamic.

The value of a field will be replaced with the encrypted value; so using the keyId as the encryption value isn’t allowed.

Pattern	Replaced by
{{record.topic}}	Name of the topic you’re encrypting the data of.
{{record.key}}	Key of the encrypted record.
{{record.value.someValueFieldName}}	Value of the field called someValueFieldName If you’re doing field-level encryption, please ensure that someValueFieldName is not included in the fields to encrypt. Otherwise, you will not be able to decrypt it.
{{record.value.someList[0].someValueField}}	Value of the field called someValueFieldName, in the first element of the list someList
{{record.header.someHeader}}	Value of the header called someHeader

Here’s a record example:

# Header
someHeader=myHeader

# Key
myKey

# Value
{
  "someValueFieldName": "I",
  "someList": [{ "someValueField": "love" }, { "someValueField": "Kafka" }]
}

Both fields take the same templates. Where the resulting key lives depends on how the Interceptor reaches its KMS:

External connection

{
  "kmsConnectionName": "PROD_VAULT",
  "recordValue": {
    "fields": [{ "fieldName": "email", "keyId": "{{record.topic}}-{{record.header.someHeader}}-{{record.key}}" }]
  }
}

The connection already knows the provider, so this creates a key called myTopic-myHeader-myKey in that KMS.

Inline configuration

{
  "kmsConfig": { "vault": { "uri": "https://vault:8200", "token": "${VAULT_TOKEN}" } },
  "recordValue": {
    "fields": [{ "fieldName": "email", "keySecretId": "vault-kms://https://vault:8200/transit/keys/{{record.topic}}-{{record.header.someHeader}}-{{record.key}}" }]
  }
}

The vault-kms:// prefix is what puts the key in Vault. Drop it — "keySecretId": "{{record.topic}}-{{record.header.someHeader}}-{{record.key}}" — and the key is created in memory instead.

KMS integration

An encryption entry names its key in one of two ways, depending on how its Interceptor reaches the KMS:

The Interceptor sets	Name each key with	Format	Example
kmsConnectionName — a KMS external connection	keyId	Bare key name; the connection already knows the provider	email-key, or alias/email-key for AWS KMS
kmsConfig — inline configuration	keySecretId	Full key URI, prefixed with the KMS type	vault-kms://vault.internal:8200/transit/keys/email-key

Set one or the other on an entry, never both. Both accept mustache templates.

External connection

{
  "kmsConnectionName": "PROD_VAULT",
  "recordValue": {
    "fields": [{ "fieldName": "email", "keyId": "email-key" }]
  }
}

Inline configuration

{
  "kmsConfig": { "vault": { "uri": "https://vault.internal:8200", "token": "${VAULT_TOKEN}" } },
  "recordValue": {
    "fields": [{ "fieldName": "email", "keySecretId": "vault-kms://vault.internal:8200/transit/keys/email-key" }]
  }
}

For the kmsType:// prefixes and their URI formats, see configure a KMS inline.

Every entry has to carry the field matching the Interceptor’s own choice, and carry it non-blank. Gateway rejects the rest when you create or update the Interceptor, naming which field to use.

Supported algorithms

• AES128_GCM (default)
• AES128_EAX
• AES256_EAX
• AES128_CTR_HMAC_SHA256
• AES256_CTR_HMAC_SHA256
• CHACHA20_POLY1305
• XCHACHA20_POLY1305
• AES256_GCM

Choosing an encryption algorithm

Gateway supports multiple encryption algorithms, with AES128_GCM as the default. When selecting an algorithm, consider your security requirements, performance needs, and message volume.

Default algorithm: AES128_GCM

AES128_GCM is the default algorithm and is suitable for most use cases. However, it has an important security limitation:

Security consideration: When the same DEK (Data Encryption Key) is used to encrypt approximately 4 billion (2³²) messages, AES-GCM’s security guarantees degrade due to nonce collision risks. According to NIST Special Publication 800-38D (Section 8.3), nonce collisions may expose encryption keys, compromising the confidentiality and integrity of data encrypted with that key.

For high-traffic scenarios, this threshold can be reached quickly. For example:

• At 10-50MB/s with 1KB message sizes, the 4-billion-message threshold can be reached in just over 24 hours
• Each unique keySecretId uses its own DEK, so the limit applies per key, not globally

If you expect a single DEK to encrypt more than ~2³² messages, consider:

• Using a different algorithm (see recommendations below)
• Implementing DEK rotation before reaching the threshold
• Using multiple keySecretId values to distribute the message count across multiple DEKs

When to keep the default (AES128_GCM)

• Low to moderate message volume per DEK (well below 2³² messages per key)
• Need for compatibility with existing AES-GCM implementations
• Hardware acceleration (AES-NI) is available, providing good performance

If you expect a single DEK to encrypt more than ~2³² messages, consider using alternative algorithms or implementing DEK rotation. Consult your cryptographic library documentation and security requirements to choose the appropriate algorithm for your use case.

Key rotation

Gateway uses envelope encryption with two types of keys: DEK (Data Encryption Key) and KEK (Key Encryption Key). Understanding how and when to rotate these keys is important for maintaining security.

KEK rotation

Customer responsibility: KEK (Key Encryption Key) rotation must be performed by the customer and is not handled automatically by Gateway. This is the customer’s responsibility and must be done through your KMS provider (AWS KMS, Azure Key Vault, HashiCorp Vault, GCP KMS, etc.).

When to rotate KEK

You should rotate your KEK based on:

• Security policies: Follow your organization’s key rotation policies and compliance requirements
• DEK encryption frequency: If DEKs are being encrypted frequently (high message volume), consider more frequent KEK rotation
• Security incidents: Rotate immediately if a KEK is suspected to be compromised
• Best practices: Many organizations rotate KEKs annually or quarterly, but the frequency should match your security requirements

After rotating a KEK, Gateway will automatically use the new KEK version for encrypting new DEKs. However, existing EDEKs encrypted with the old KEK version will still be de-cryptable as long as the old KEK version remains available in your KMS.

Most KMS providers retain old key versions for backward compatibility, allowing you to decrypt historical data while new data uses the rotated key.

Supported compression types

• none
• gzip
• snappy
• lz4
• zstd

Encryption error policy

This policy determines the actions when an encryption Interceptor encounters a record that’s already encrypted.

Error policy	Description
fail_on_encrypted	The encryption operation will fail with an exception when encountering already encrypted records. This is the default behavior and maintains backward compatibility.
skip_already_encrypted	The encryption Interceptor will skip already encrypted records and pass them through unchanged. This enables chaining multiple encryption Interceptors together.

Example configuration with error policy:

{
  "name": "mySchemaBasedEncryptPlugin",
  "pluginClass": "io.conduktor.gateway.interceptor.EncryptSchemaBasedPlugin",
  "config": {
    "topic": "sensitive-data",
    "defaultKeySecretId": "vault-kms://vault:8200/transit/keys/default-key",
    "defaultAlgorithm": "AES256_GCM",
    "tags": ["PII", "ENCRYPT"],
    "errorPolicy": "skip_already_encrypted",
    "kmsConfig": {
      "vault": {
        "uri": "http://vault:8200",
        "token": "${VAULT_TOKEN}"
      }
    }
  }
}

Tokenization

Tokenization replaces a sensitive value with an unrelated token, and keeps the mapping in Vault rather than in your Kafka record. A team can work with the tokenized field without the values ever leaving Vault.

Tokenization only works with the Transform Secrets Engine, available in Vault Enterprise. No other KMS provider supports it for now, but we would love to hear from you.

Gateway calls the Transform role named in your keySecretId and writes the token it gets back into the record. Vault owns the behavior, so create the role with the tokenization transformation, which is the configuration Gateway is tested against:

vault write transform/transformations/tokenization/email allowed_roles=email
vault write transform/role/email transformations=email

With this transformation, Vault stores a mapping of the token to the encrypted plaintext, and the token carries no link to the original value, so it isn’t format-preserving.

When to use tokenization

Tokenization fits when a team has to work with the data rather than just move it: reporting on unique customer counts, or joining datasets on a customer reference.

It’s a weaker fit when the same clear text can reach the same place as the tokens. Tokens are deterministic, so anyone holding a clear value and its token learns every other record carrying that value. If you can’t guarantee clear text stays out, use non-deterministic field level encryption instead. See choose how to protect your data for the full comparison.

Configure tokenization

Tokenization uses the existing encryption Interceptors (EncryptPlugin, DecryptPlugin and so on). To tokenize data, use the encryption Interceptors with a vault-transform:// prefix in your keySecretId. To de-tokenize and retrieve the original values, use the decryption Interceptors with the same configuration. See how to configure Vault.

Here’s a sample field configured for tokenization:

fields:
  - fieldName: "email"
    keySecretId: "vault-transform://email-key"

Configure the KMS inline for tokenization. A KMS external connection only reaches Vault’s Transit engine, so it can’t use the Transform engine tokenization relies on. Set kmsConfig on the Interceptor rather than kmsConnectionName.

Decryption configuration

Now that your fields or payload are encrypted, you can decrypt them using the Interceptor DecryptPlugin.

Key	Type	Default	Description
topic	String	.*	Topics matching this regex will have the Interceptor applied.
schemaRegistryConfig	SchemaRegistry		Configuration of your schema registry, is needed if you want to decrypt into Avro, JSON or Protobuf schemas.
schemaRegistryConnectionName	String		Name of a schema registry external connection. An alternative to schemaRegistryConfig — set one or the other, not both.
kmsConfig	KMS		Configuration of one or multiple KMS
kmsConnectionName	String		Name of a KMS external connection. An alternative to kmsConfig — set one or the other, not both
recordValueFields	List[String]		Only for field-level encryption - List of fields to decrypt in the value. If empty, we decrypt all the encrypted fields.
recordKeyFields	List[String]		Only for field-level encryption - List of fields to decrypt in the key. If empty, we decrypt all the encrypted fields.
recordHeaderFields	List[String]		Only for field-level encryption - List of headers to decrypt. If empty, we decrypt all the encrypted headers.
enableAuditLogOnError	Boolean	true	The audit log will be enabled when an error occurs during encryption/decryption
errorPolicy	String	return_encrypted	Determines the action if there is an error during decryption. The options are return_encrypted, fail_fetch and crypto_shred_safe_fail_fetch See Decryption Error Policy section for more details.
throttleTimeMs	Integer	0	When decryption fails, apply client throttling for the specified time in milliseconds. This helps prevent the overwhelming of the system during decryption failures. Find out more about client throttling

Decryption error policy

This policy determines the action if there is an error during decryption.

Error policy	Description
return_encrypted	The encrypted data is returned to the client.
fail_fetch	The client will receive an error for the fetch and no data. When using the Gateway KMS, data that fails to decrypt because no key was found will use the policy return_encrypted instead (supports crypto shredding for just this KMS system).
crypto_shred_safe_fail_fetch	The client will receive an error for the fetch and no data. Data that fails to decrypt because no key was found will use the policy return_encrypted instead (supports crypto shredding for all KMS systems).

KMS configuration

An encrypt or decrypt Interceptor reaches its KMS in one of two ways:

Path	The Interceptor sets	Where the provider is configured	Keys are named with
Name a KMS external connection	kmsConnectionName	Once, in Gateway’s environment	keyId
Configure a KMS inline	kmsConfig	On each Interceptor	keySecretId

Set one or the other, never both — Gateway rejects a configuration with both rather than silently picking one. See how each names its keys.

Configuration properties

Property	Type	Default	Description
keyTtlMs	long	3600000	Key’s time-to-live in milliseconds. The default is 1 hour. Disable the cache by setting it to 0.

Name a KMS external connection

KMS external connections are a preview feature and subject to change. Inline kmsConfig supports more providers, keeps working and isn’t deprecated — see which providers are available as an external connection.

Instead of a kmsConfig block, an encryption or decryption Interceptor can name a KMS external connection declared once when Gateway starts. The address, credentials and engine live in one place, so rotating a Vault credential is a single edit rather than one per Interceptor.

Key	Type	Description
kmsConnectionName	String	The <NAME> of a KMS connection, declared with GATEWAY_EXTERNAL_CONNECTION_<NAME>_CONFIG_<PROPERTY> variables. Set at the top level of the Interceptor configuration, not inside kmsConfig

Configure a KMS inline

Set a kmsConfig block on the Interceptor, holding one or more providers from the catalogue below. Each Interceptor carries its own copy of the address and credentials.

Each key is named with a full URI in keySecretId. Any that doesn’t match one of the schemas detailed below will be rejected and the encryption operation will fail.

Keys are strings that start with a letter followed by a combination of letters, underscores (_), hyphens (-) and numbers. Special characters are not allowed.

They also work with the mustache pattern.

To make sure the key is created in your KMS, configure the provider in kmsConfig and use the matching format:

KMS	KMS identifier prefix	Key URI format	Example
In-Memory	in-memory-kms://	in-memory-kms://<key-id>	in-memory-kms://my-password-key-id
Azure	azure-kms://	azure-kms://<scheme>://<key-vault-name>.vault.azure.net/keys/<object-name>/<object-version>	azure-kms://https://my-key-vault.vault.azure.net/keys/conduktor-gateway/4ceb7a4d1f3e4738b23bea870ae8745d
AWS	aws-kms://	aws-kms://arn:aws:kms:<region>:<account-id>:key/<key-id>	aws-kms://arn:aws:kms:us-east-1:123456789012:key/password-key-id
GCP	gcp-kms://	gcp-kms://projects/<project-id>/locations/<location-id>/keyRings/<key-ring-id>/cryptoKeys/<key-id>	gcp-kms://projects/my-project/locations/us-east1/keyRings/my-key-ring/cryptoKeys/password-key-id
Fortanix	fortanix-kms://	fortanix-kms://<key-name>	fortanix-kms://password-key-id
Gateway	gateway-kms://	gateway-kms://<key-id> (uses master key from KMS above)	gateway-kms://user-{{record.key}}
Vault (Transit)	vault-kms://	vault-kms://[scheme://]<vault-host>/transit/keys/<key-id>	vault-kms://https://vault:8200/transit/keys/password-key-id
Vault Transform	vault-transform://	vault-transform://<role-name>	vault-transform://password-role
Test tokenization	test-tokenization://	test-tokenization://<role-name>	test-tokenization://password-role

Vault appears twice because encryption and tokenization use two different HashiCorp modules. vault-kms:// uses the Transit Engine and encrypts. vault-transform:// uses the Transform Secrets Engine, available in Vault Enterprise only, and tokenizes. Having one doesn’t give you the other.

In-memory and test-tokenization modes are for testing and development purposes only. Test tokenization also requires GATEWAY_FEATURE_FLAGS_TEST_TOKENIZATION to be set to TRUE

If not specified, vault KMS scheme defaults to https. This means that vault-kms://https://vault:8200/transit/keys/password-key-id and vault-kms://vault:8200/transit/keys/password-key-id are identical.

Keys are strings that start with a letter followed by a combination of letters, underscores (_), hyphens (-) and numbers. Special characters are not allowed. Keys also work with the Mustache pattern described above.

Choose your KMS provider

Provider	When to use	Key consideration
In-memory	Local development and testing only	Keys don’t persist - data becomes unreadable after Gateway restart
Gateway	Crypto shredding with high-volume per-record keys	Stores encrypted keys locally, reducing KMS costs for millions of unique keys
AWS KMS	AWS infrastructure	Native IAM integration, no credential management needed
Azure Key Vault	Azure infrastructure	Managed identity support, integrates with Azure services
Vault KMS	Multi-cloud or on-premises	Most flexible authentication (11 methods), works anywhere
GCP KMS	Google Cloud infrastructure	Service account integration, follows GCP security model
Fortanix KMS	FIPS 140-2 compliance required	Hardware security module (HSM) backed, meets strict regulatory requirements

In-memory KMS (Development only)

In-memory KMS

This should not be used on production data.

Keys in in-memory KMS are not persisted, this means that if you do one of the following, you won’t be able to decrypt old records, losing the data.

• Use a Gateway cluster with more than a single node or
• restart Gateway or
• change the Interceptor configuration

Gateway KMS (Crypto shredding)

Gateway KMS

This KMS type is effectively a delegated storage model and is designed to support encryption use cases which generate unique secret Ids per record or even field (typically via the Mustache template support for a secret Id). This technique is used in crypto-shredding type scenarios e.g. encrypting records per user with their own key.

It provides the option to leverage your KMS for security via a single master key, but efficiently and securely store many per-record level encryption keys (DEKs) in the Gateway managed store. For some architectures this can provide performance and cost savings for encryption use cases which generate a high volume of secret key Ids.

Key	Type	Description
masterKeyId	String	The master key secret Id used to encrypt any keys stored in the Gateway managed storage. This is in the same format as the keySecretId that’s used for encryption and the valid values are the same.
maxKeys	Number	The maximum number of secret Id references to be cached in memory for re-use. To avoid creating new encryption keys (DEKs), this needs to be larger than the total number of expected secret Ids. By default, it’s the same as maxKeys in cache config or 1000,000, if maxKeys isn’t set.

The masterKeyId is used to secure every key for this configuration, stored by Gateway. Find out more about the secret key formats. You have to also supply a valid configuration for the KMS type referenced by the master key so this can be used.

If this key is dropped from the backing KMS, then all keys stored by Gateway for that master key will become unreadable.

Gateway KMS encryption example

Here’s a sample configuration for the Gateway KMS using a Vault-based master key:

"kmsConfig": {
   "gateway": {
      "masterKeyId": "vault-kms://vault:8200/transit/keys/applicants-1-master-key".
      "maxKeys" : 10000000
   },
   "vault": {
      "uri": "https://vault:8200",
      "token": "my-vault-token",
      "trustStore": {
        "trustStorePath": "/security/truststore.jks"
      }
   }
}

This can then be used to encrypt a field using gateway-kms:// as the secret key type:

"recordValue": {
   "fields": [
      {
         "fieldName": "name",
         "keySecretId": "gateway-kms://fieldKeySecret-name-{{record.key}}"
      }
   ]
}

When processing a record for the first time using this configuration, Gateway will:

1. generate a DEK to encrypt the field data,
2. turn it into an EDEK by encrypting with the masterKeyId secret from vault and
3. store the EDEK in Gateway storage.

If a record key was 123456, the associated EDEK would be stored on a kafka record with the following key:

{"algorithm":"AES128_GCM","keyId":"gateway-kms://fieldKeySecret-name-123456","uuid":"<UNIQUE_PER_EDEK_GENERATED>"}

Multiple records produced against this config would cause multiple EDEKs to be saved in the Gateway storage (due to the {{record.key}} template giving a unique key for each Kafka record key).

If there are multiple Gateway nodes running, it’s also possible for multiple DEKs/EDEKs to be generated for the same record key. Two nodes processing different records with the same record key at the same time could both assume they were generating a DEK/EDEK for the first time. In this scenario, there would be two EDEKs in the Gateway storage with the same keyId but they would each have a different UUID.

For example:

{"algorithm":"AES128_GCM","keyId":"gateway-kms://fieldKeySecret-name-123456","uuid":"2cd8125a-b55f-4214-a528-be3c9b47519b"}
{"algorithm":"AES128_GCM","keyId":"gateway-kms://fieldKeySecret-name-123456","uuid":"d8fcccf3-8480-4634-879a-48deed4e0e72"}

Nonetheless, there will only ever be one master key stored in the vault KMS, which is used to encrypt every DEK.

This feature provides flexibility for your KMS storage and key management setups - and is particularly useful for high volume crypto shredding.

Decryption using Gateway KMS

When using the gateway-kms secret key Id type, the decryption configuration used to decrypt the data has to also specify the masterKeyId, so that it can securely decrypt the keys stored in the local Gateway storage.

Here’s a sample setup:

"config": {
   "topic": "secure-topic",
   "kmsConfig": {
      "gateway": {
         "masterKeyId": "vault-kms://vault:8200/transit/keys/secure-topic-master-key"
      },
      "vault": {
         "uri": "https://vault:8200",
         "token": "my-token-for-vault",
         "trustStore": {
           "trustStorePath": "/security/truststore.jks"
         }
      }
   }
}

Crypto shredding

When using the gateway-kms secret key Id type, you can efficiently crypto shred EDEKs in the Gateway storage, so that anyone using the decryption plugin will immediately lose access to the associated encrypted data.

To do this, scan the Gateway storage Kafka topic (by default, _conduktor_gateway_encryption_keys) for every message matching the associated qualified secret Id.

For example, a qualified secretId of gateway-kms://fieldKeySecret-name-123456 might have the following keys:

{"algorithm":"AES128_GCM","keyId":"gateway-kms://fieldKeySecret-name-123456","uuid":"2cd8125a-b55f-4214-a528-be3c9b47519b"}
{"algorithm":"AES128_GCM","keyId":"gateway-kms://fieldKeySecret-name-123456","uuid":"d8fcccf3-8480-4634-879a-48deed4e0e72"}

Publishing a message for each of these keys back to the same topic with a value of null (i.e. a tombstone) will effectively perform Crypto Shredding.

This process won’t prevent the creation of new keys if new messages are sent using the same record key; it only ensures that messages using the crypto shredded keys remain unrecoverable.

AWS KMS

AWS KMS

To set your AWS KMS, include this section in your Interceptor config, below aws.

You can use one of these two authentication methods:

• basic authentication or
• session.

Make sure to follow the right method and provided the correct properties. Omit both and Gateway falls back to the AWS SDK’s default credentials provider.

Naming keys through an external connection. An AWS KMS external connection carries the region and account, so each keyId is a bare key name rather than an ARN. Gateway builds the ARN for you:

arn:<partition>:kms:<region>:<accountId>:key/<keyId>

• The partition comes from the region, so GovCloud, China and the ISO regions work without naming one.
• An alias/<name> id is used as the ARN resource as-is, so it becomes …:<accountId>:alias/<name> rather than gaining a key/ prefix.
• Anything else is rejected when you create or update the Interceptor, with AWS keyId '…' must be a key id or 'alias/<name>'; the region and account are part of connection <NAME>. Pasting a full ARN from the console is the usual cause.

For enhanced security, you can hide the sensitive values using environment variables as secrets.

Key	Type	Description
Basic authentication
basicCredentials.accessKey	String	Access key.
basicCredentials.secretKey	String	Secret key.
Session
sessionCredentials.accessKey	String	Access key.
sessionCredentials.secretKey	String	Secret key.
sessionCredentials.sessionToken	String	Session token.
Managed identity		Configure the KMS from the context, not using variables. This will be overwritten if a specific KMS is configured within the Interceptor.

Azure Key Vault

Azure KMS

To set your Azure KMS, include this section in your Interceptor config, below azure.

You can use one of these two authentication methods:

• token or
• username and password.

Make sure you’ve followed the right method, and that you’ve provided the correct properties.

For enhanced security, you can hide the sensitive values using environment variables as secrets.

Key	Type	Description
Token
tokenCredential.clientId	string	Client ID.
tokenCredential.tenantId	string	Tenant ID.
tokenCredential.clientSecret	string	Client secret.
Username and password
usernamePasswordCredential.clientId	string	Client ID.
usernamePasswordCredential.tenantId	string	Tenant ID.
usernamePasswordCredential.username	string	Username.
usernamePasswordCredential.password	string	Password.
Managed identity		Configure the KMS from the context, and not using variables. This will be overwritten if a specific KMS is configured within the Interceptor.

Fortanix KMS

Fortanix KMS

To set your Fortanix KMS, include this section in your Interceptor config under fortanix:

Key	Type	Description
type	String	Authentication method: API_KEY or USERNAME_PASSWORD
basePath	String	Fortanix DSM base URL: https://support.fortanix.com/docs/fortanix-dsm-saas-global-availability-map
apiKey	String	type is API_KEY: Fortanix API key. Gateway only supports AES type keys.
username	String	type is USERNAME_PASSWORD: Fortanix username
password	String	type is USERNAME_PASSWORD: Fortanix password

Alternatively, you can use the environment variables for sensitive credentials and configuration values. Set them in your Gateway deployment and they will be resolved at runtime. In that case, you don’t have to supply a Fortanix {} block in the KMS config.

If the specified encryption key doesn’t exist in Fortanix DSM, Gateway will automatically create it with the following configuration:

• Key type: AES symmetric key
• Mode: CBC
• Key Size: 256
• Permissions: ENCRYPT, DECRYPT

Google Cloud Platform KMS

Google Cloud Platform KMS

To set your Google Cloud Platform (GCP) KMS, include this section in your Interceptor config, under gcp:

You must first configure the service account key file.

For enhanced security, you can hide the sensitive values using environment variables as secrets.

Key	Type	Description
serviceAccountCredentialsFilePath	String	Service account key file in GCP.
enableDetailedErrorLogging	Boolean	Use with caution - error responses may contain sensitive information. Defines whether to enable detailed GCP KMS error response logging. When enabled, full error responses from GCP KMS will be logged. When disabled (default), only safe fields (code, status, reason, domain) will be logged.
Managed identity		Configure the KMS from the context, and not using variables. This will be overwritten if a specific KMS is configured within the Interceptor.

Vault KMS

Vault KMS

Gateway supports two data security backends for Vault:

• Transit Secrets Engine - the default and most useful
• Transform Secrets Engine - supports non-standard features like tokenization but requires Vault Enterprise account)

To configure Vault, add a new Vault section in your Interceptor configuration. For enhanced security, you can hide the sensitive values using the environment variables as secrets.

Key	Type	Description
uri	String	Vault server base URI. Example: vault:8200 (assumed https for backward compatibility) or https://vault:8200 or http://vault:8200.
namespace	String	Namespace.
type	String	Required for all types of VaultKMSConfig. Determines the type of authentication to use. Supported types: -TOKEN -USERNAME_PASSWORD -GITHUB -LDAP -APP_ROLE -KUBERNETES -GCP -AWS_EC2_PKCS7 -AWS_EC2 -AWS_IAM -JWT
trustStore	TrustStore	Trust store configuration (JKS format) for verifying the Vault server’s TLS certificate when using HTTPS with a privately-signed or internal certificate authority.
keyStore	KeyStore	Key store configuration (JKS format) containing the client certificate and private key for mutual TLS (mTLS) authentication when the Vault server requires client certificate verification.
Managed identity		Load authentication information from the below environment variables.
VAULT_URI		Vault server base URI.
VAULT_NAMESPACE		Vault namespace.

TrustStore

Key	Type	Description	Environment variable
trustStorePath	String	Path to JKS trust store file containing the Certificate Authority certificate.	VAULT_SSL_TRUST_STORE_PATH
trustStorePassword	String	Password for the JKS trust store file containing the Certificate Authority certificate.	VAULT_SSL_TRUST_STORE_PASSWORD

When using HTTPS Vault URIs without an explicitly configured trust store, Gateway will attempt to use the VAULT_SSL_TRUST_STORE_PATH and VAULT_SSL_TRUST_STORE_PASSWORD environment variables if available.

KeyStore

Key	Type	Description	Environment variable
keyStorePath	String	Path to JKS key store file containing the client certificate.	VAULT_SSL_KEY_STORE_PATH
keyStorePassword	String	Password for the JKS key store file containing the client certificate.	VAULT_SSL_KEY_STORE_PASSWORD

The Vault client library requires that JKS key stores contain both a private key and client certificate with identical passwords. When using HTTPS Vault URIs without an explicitly configured key store, Gateway will attempt to use the VAULT_SSL_KEY_STORE_PATH and VAULT_SSL_KEY_STORE_PASSWORD environment variables if available.

Vault authentication types

Key	Type	Description
Token authentication		Use token authentication.
type	String	Must be TOKEN. Indicates the type of authentication.
token	String	Security token for accessing Vault.
Managed identity		Load authentication information from the below environment variables.
VAULT_AUTH_TYPE	String	Must be TOKEN. Indicates the type of authentication.
VAULT_TOKEN		Token to use for accessing Vault.
Username and password		Use username and password authentication.
type	String	Must be USERNAME_PASSWORD. Indicates the type of authentication.
username	String	Username for accessing Vault.
password	String	Password for accessing Vault.
userpassAuthMount	String	(Optional) Mount path for the userpass auth method.
Managed identity		Load authentication information from the below environment variables.
VAULT_AUTH_TYPE	String	Must be USERNAME_PASSWORD. Indicates the type of authentication.
VAULT_USERNAME		Username for accessing Vault.
VAULT_PASSWORD		Password for accessing Vault.
VAULT_AUTH_MOUNT		(Optional) Mount path for the userpass auth method.
GitHub authentication		Use GitHub token authentication.
type	String	Must be GITHUB. Indicates the type of authentication.
token	String	GitHub personal access token.
githubAuthMount	String	(Optional) Mount path for the GitHub auth method.
Managed identity		Load authentication information from the below environment variables.
VAULT_AUTH_TYPE	String	Must be GITHUB. Indicates the type of authentication.
VAULT_GITHUB_TOKEN		GitHub token for accessing Vault.
VAULT_AUTH_MOUNT		(Optional) Mount path for the GitHub auth method.
LDAP authentication		Use LDAP authentication.
type	String	Must be LDAP. Indicates the type of authentication.
username	String	LDAP username.
password	String	LDAP password.
ldapAuthMount	String	(Optional) Mount path for the LDAP auth method.
Managed identity		Load authentication information from the below environment variables.
VAULT_AUTH_TYPE	String	Must be LDAP. Indicates the type of authentication.
VAULT_LDAP_USERNAME		LDAP username.
VAULT_LDAP_PASSWORD		LDAP password.
VAULT_AUTH_MOUNT		(Optional) Mount path for the LDAP auth method.
AppRole authentication		Use AppRole authentication.
type	String	Must be APP_ROLE. Indicates the type of authentication.
roleId	String	Role ID for AppRole authentication.
secretId	String	Secret ID for AppRole authentication.
path	String	(Optional) Mount path for the AppRole auth method.
Managed identity		Load authentication information from the below environment variables.
VAULT_AUTH_TYPE	String	Must be APP_ROLE. Indicates the type of authentication.
VAULT_APP_ROLE_ID		Role ID for AppRole authentication.
VAULT_APP_SECRET_ID		Secret ID for AppRole authentication.
VAULT_APP_PATH		(Optional) Mount path for the AppRole auth method.
Kubernetes authentication		Use Kubernetes authentication.
type	String	Must be KUBERNETES. Indicates the type of authentication.
role	String	Kubernetes role.
path	String	(Optional) Token file path for the Kubernetes auth method.
Managed identity		Load authentication information from the below environment variables.
VAULT_AUTH_TYPE	String	Must be KUBERNETES. Indicates the type of authentication.
VAULT_KUBERNETES_ROLE		Kubernetes role.
VAULT_KUBERNETES_PATH		(Optional) Token file path for the Kubernetes auth method.
GCP authentication		Use Google Cloud Platform authentication.
type	String	Must be GCP. Indicates the type of authentication.
role	String	GCP role for authentication.
jwt	String	JWT token issued by Google Cloud Platform.
Managed identity		Load authentication information from the below environment variables.
VAULT_AUTH_TYPE	String	Must be GCP. Indicates the type of authentication.
VAULT_GCP_ROLE		GCP role for authentication.
VAULT_GCP_JWT		JWT token for accessing Vault.
AWS EC2 authentication (PKCS7)		Use AWS EC2 PKCS7 authentication.
type	String	Must be AWS_EC2_PKCS7. Indicates the type of authentication.
role	String	AWS role for EC2 authentication.
pkcs7	String	PKCS7 identity document.
nonce	String	(Optional) Nonce value for EC2 authentication.
Managed identity		Load authentication information from the below environment variables.
VAULT_AUTH_TYPE	String	Must be AWS_EC2_PKCS7. Indicates the type of authentication.
VAULT_AWS_ROLE		AWS role for EC2 authentication.
VAULT_AWS_PKCS7		PKCS7 identity document.
VAULT_AWS_NONCE		(Optional) Nonce value for EC2 authentication.
VAULT_AUTH_MOUNT		(Optional) Mount path for the AWS EC2 PKCS7 auth method.
AWS EC2 authentication		Use AWS EC2 identity authentication.
type	String	Must be AWS_EC2. Indicates the type of authentication.
role	String	AWS role for EC2 authentication.
identity	String	AWS identity document.
signature	String	AWS signature for authentication.
nonce	String	Nonce value for EC2 authentication.
Managed identity		Load authentication information from the below environment variables.
VAULT_AUTH_TYPE	String	Must be AWS_EC2. Indicates the type of authentication.
VAULT_AWS_ROLE		AWS role for EC2 authentication.
VAULT_AWS_IDENTITY		AWS identity document.
VAULT_AWS_SIGNATURE		AWS signature for authentication.
VAULT_AWS_NONCE		(Optional) Nonce value for EC2 authentication.
VAULT_AUTH_MOUNT		(Optional) Mount path for the AWS EC2 auth method.
AWS IAM authentication		Use AWS IAM Authentication.
type	String	Must be AWS_IAM. Indicates the type of authentication.
role	String	AWS role for IAM authentication.
iamRequestUrl	String	IAM request URL for authentication.
iamRequestBody	String	IAM request body for authentication.
iamRequestHeaders	String	IAM request headers for authentication.
Managed identity		Load authentication information from the below environment variables.
VAULT_AUTH_TYPE	String	Must be AWS_IAM. Indicates the type of authentication.
VAULT_AWS_ROLE		AWS role for IAM authentication.
VAULT_AWS_IAM_REQUEST_URL		IAM request URL for authentication.
VAULT_AWS_IAM_REQUEST_BODY		IAM request body for authentication.
VAULT_AWS_IAM_REQUEST_HEADERS		IAM request headers for authentication.
JWT authentication		Use JWT authentication.
type	String	Must be JWT. Indicates the type of authentication.
jwt	String	JWT token for authentication.
provider	String	JWT provider for authentication.
role	String	JWT role for authentication.
Managed identity		Load authentication information from the below environment variables.
VAULT_AUTH_TYPE	String	Must be JWT. Indicates the type of authentication.
VAULT_JWT		JWT token for authentication.
VAULT_JWT_PROVIDER		JWT provider for authentication.

Example:

{
  "type": "APP_ROLE",
  "uri": "https://vault.example.com",
  "roleId": "my-role-id",
  "secretId": "my-secret-id",
  "trustStore": {
    "trustStorePath": "/security/truststore.jks"
  }
}

Connection backoff

If there’s a connection failure (e.g., a network partition error), Gateway will automatically keep trying to reconnect to Vault.

You can adjust this by modifying the kmsConfig.vault.connectionBackoff object:

Key	Type	Default	Description
backoffDelay	Integer	5	Initial retry delay for Vault health checks
backoffMaxDelay	Integer	30	Maximum retry delay
backoffChronoUnit	String	SECONDS	Time unit for delays (SECONDS, MILLIS, etc.)
backoffDelayFactor	Double	1.1	Exponential backoff multiplier

Example:

{
  "type": "TOKEN",
  "uri": "https://vault:8200",
  "token": "${VAULT_TOKEN}",
  "connectionBackoff": {
    "backoffDelay" : 5,
    "backoffMaxDelay": 30,
    "backoffChronoUnit" : "SECONDS",
    "backoffDelayFactor" : 1.1
  }
}

Configuring Vault Transit Secrets Engine

Here’s the minimum Vault policy required for encryption and decryption to work with vault:// prefixed keys:

path "transit/*" {
  capabilities = ["read", "update"]
}

Configuring Vault Transform Secrets Engine

Here’s the minimum Vault policy required for encryption and decryption to work with vault-transform:// prefixed keys,

path "transform/*" {
  capabilities = ["read", "update"]
}

When using the Vault transform secrets engine, you can configure additional caching using the kmsConfig.vault.transformEngineCache object:

Key	Type	Default	Description
enabled	Boolean	false	Whether to enable caching
populateOnSecure	Boolean	false	Cache tokens while securing, not just de-securing
ttlMs	Long	600000	Token cache TTL in milliseconds after last access (1 hour default)
maxSize	Long	1000000	Maximum number of tokens to cache (1 million default).

Example:

{
  "type": "TOKEN",
  "uri": "https://vault:8200",
  "token": "${VAULT_TOKEN}",
  "transformEngineCache": {
    "enabled" : true,
    "populateOnSecure": false,
    "ttlMs": 600000,
    "maxSize": 1000000
  }
}

Client throttling configuration

When encryption or decryption operations fail, you can configure client throttling to help protect your system from being overwhelmed during error conditions.

To enable client throttling, set the throttleTimeMs parameter in your encryption/decryption Interceptor config:

• throttleTimeMs = 0 (default): no throttling - clients receive immediate error responses
• throttleTimeMs > 0: clients will be throttled for the specified time in milliseconds when operations fail

When throttleTimeMs is configured with a value greater than 0:

• cluster stability is protected: Gateway automatically throttles clients to prevent system overload
• client compliance is built-in: Kafka clients automatically pause for the specified throttle time between requests
• failure cascades are prevented: throttling reduces retry pressure, allowing brokers to recover

Configuration example:

{
  "name": "myEncryptPlugin",
  "pluginClass": "io.conduktor.gateway.interceptor.EncryptPlugin",
  "config": {
    "throttleTimeMs": 5000
  }
}