> ## Documentation Index
> Fetch the complete documentation index at: https://docs.conduktor.io/llms.txt
> Use this file to discover all available pages before exploring further.
# 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](#audit-interceptor) |
| Hide sensitive fields from consumers | [Data masking](#data-masking-interceptor) |
| Add metadata to messages | [Header injection](#dynamic-header-injection-interceptor) |
| Sign and verify message integrity | [Message integrity Interceptor](#message-integrity-interceptor) |
| Encrypt data | [Encryption](#encryption-interceptors) |
| Keep data usable for analytics while protecting it | [Tokenization](#tokenization) |
| Delete a single customer's data from Kafka | [Crypto shredding](/guide/tutorials/crypto-shredding) |
**From our blog:** [FedRAMP High for Kafka without replatforming](https://www.conduktor.io/blog/fedramp-high-for-kafka-without-replatforming) Meet FedRAMP High encryption requirements on your existing Kafka, no replatforming.
**[Kafka and HIPAA 2026](https://www.conduktor.io/blog/kafka-hipaa-compliance)** 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](https://www.confluent.io/blog/schema-id-kafka-headers-data-governance), require **Gateway 3.21.0** or later.
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](/guide/conduktor-in-production/deploy-artifacts/deploy-gateway/external-connections#connect-to-a-schema-registry), declared once when Gateway starts with variables that follow this pattern:
```
GATEWAY_EXTERNAL_CONNECTION__CONFIG_
```
| Key | Type | Description |
| ------------------------------ | ------ | ----------------------------------------------------------------------------------------------------- |
| `schemaRegistryConnectionName` | string | The `` 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:
```json theme={null}
"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](/guide/conduktor-in-production/deploy-artifacts/deploy-gateway/external-connections#what-happens-at-startup).
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](/guide/reference/data-security#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](https://docs.aws.amazon.com/sdk-for-java/v1/developer-guide/credentials.html#credentials-default) ) and the credentials required can be passed this way to Gateway as part of its core configuration.
[Read our blog about schema registry](https://www.conduktor.io/blog/what-is-the-schema-registry-and-why-do-you-need-to-use-it/).
## 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](/guide/conduktor-in-production/deploy-artifacts/deploy-gateway/environment-variables#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
```bash theme={null}
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
]
}
}
}'
```
```yaml theme={null}
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:
```bash theme={null}
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](#schema-registry) | | The schema registry in use. Required for Avro, JSON Schema or Protobuf data. |
| schemaRegistryConnectionName | String | | Name of a [schema registry external connection](/guide/conduktor-in-production/deploy-artifacts/deploy-gateway/external-connections). 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](#encryption-configuration).
Field level encryption is not affected by this behavior.
### Data masking Interceptor example
```bash theme={null}
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"
]
}
]
}
}
}'
```
```yaml theme={null}
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:
```bash theme={null}
conduktor apply -f data-masking-interceptor.yaml
```
#### Secured schema registry
```bash theme={null}
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"
]
}
]
}
}
}'
```
```yaml theme={null}
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:
```bash theme={null}
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](/guide/conduktor-in-production/deploy-artifacts/deploy-gateway/load-balancing#capturing-the-client-ip-address). 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](#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](#error-handling). |
| schemaRegistryConfig | [Schema registry](#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](/guide/conduktor-in-production/deploy-artifacts/deploy-gateway/external-connections). 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
```bash theme={null}
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"
}
}
}
}'
```
```yaml theme={null}
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:
```bash theme={null}
conduktor apply -f header-injection-interceptor.yaml
```
Let's produce a simple record to the `injectHeaderTopic` topic.
```bash theme={null}
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`.
```bash theme={null}
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
```md theme={null}
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 `/data/#`.
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](#authenticate-with-vault-for-message-integrity) 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](/guide/conduktor-in-production/admin/audit-logs) and [Environment variables](/guide/conduktor-in-production/deploy-artifacts/deploy-gateway/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](#authenticate-with-vault-for-message-integrity) 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 `` 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
```yaml theme={null}
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
```
```yaml theme={null}
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](/guide/use-cases/kafka-encryption)
* [encryption examples](/guide/tutorials/configure-encryption)
### 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](#schema-registry) | | 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](/guide/conduktor-in-production/deploy-artifacts/deploy-gateway/external-connections). 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](#kms-configuration) | | Configuration of one or multiple KMS. |
| `kmsConnectionName` | String | | Name of a [KMS external connection](#name-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](#client-throttling) |
| `errorPolicy` | [String](#encryption-error-policy) | `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**](#list-based) | | | |
| `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**](#schema-based) | | | |
| `defaultKeySecretId` | [Key reference](#kms-integration) | | 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](#kms-integration) | | Bare key name for every encrypted field, when the Interceptor names a KMS connection. |
| `defaultAlgorithm` | [Algorithm](#supported-algorithms) | `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](#kms-integration) | | Key URI, when the Interceptor configures its KMS inline. Can be a template for Crypto Shredding use cases. |
| `payload.keyId` | [Key reference](#kms-integration) | | Bare key name, when the Interceptor names a KMS connection. Set this or `keySecretId`, never both. |
| `payload.algorithm` | [Algorithm](#supported-algorithms) | `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](#kms-integration) | | Key URI, when the Interceptor configures its KMS inline. Can be a template for crypto shredding use cases. |
| `fields[].keyId` | [Key reference](#kms-integration) | | Bare key name, when the Interceptor names a KMS connection. Set this or `keySecretId`, never both. |
| `fields[].algorithm` | [Algorithm](#supported-algorithms) | | Algorithm to use to encrypt this field. |
[Check out the encryption examples](/guide/tutorials/configure-encryption#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](#kms-integration) | | Key URI, when the Interceptor configures its KMS inline. Can be a template for crypto shredding use cases. |
| `payload.keyId` | [Key reference](#kms-integration) | | Bare key name, when the Interceptor names a KMS connection. Set this or `keySecretId`, never both. |
| `payload.algorithm` | [Algorithm](#supported-algorithms) | `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](#kms-integration) | | Key URI, when the Interceptor configures its KMS inline. Can be a template for crypto shredding use cases. |
| `fields[].keyId` | [Key reference](#kms-integration) | | Bare key name, when the Interceptor names a KMS connection. Set this or `keySecretId`, never both. |
| `fields[].algorithm` | [Algorithm](#supported-algorithms) | | |
| **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](/guide/tutorials/configure-encryption#encryption-examples).
#### 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](#kms-integration) | | Key URI, when the Interceptor configures its KMS inline. Can be a template for crypto shredding use cases. |
| `conduktor.algorithm` | [Algorithm](#supported-algorithms) | `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](#name-a-kms-external-connection) doesn't yet honor per-field keys from a schema.
[Check out the encryption example](/guide/tutorials/configure-encryption#encryption-examples).
#### 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](#name-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:
```md theme={null}
# 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:
```json theme={null}
{
"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.
```json theme={null}
{
"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](#name-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](#configure-a-kms-inline) | `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](#secret-keys).
```json theme={null}
{
"kmsConnectionName": "PROD_VAULT",
"recordValue": {
"fields": [{ "fieldName": "email", "keyId": "email-key" }]
}
}
```
```json theme={null}
{
"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](#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)](https://csrc.nist.gov/pubs/sp/800/38/d/final), 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:**
```json theme={null}
{
"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](https://developer.hashicorp.com/vault/docs/secrets/transform), 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:
```bash theme={null}
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](#encryption-configuration) instead. See [choose how to protect your data](/guide/use-cases/kafka-encryption) 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](#vault-kms).
Here's a sample field configured for tokenization:
```yaml theme={null}
fields:
- fieldName: "email"
keySecretId: "vault-transform://email-key"
```
Configure the KMS inline for tokenization. A [KMS external connection](#name-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](#schema-registry) | | 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](/guide/conduktor-in-production/deploy-artifacts/deploy-gateway/external-connections). An alternative to `schemaRegistryConfig` — set one or the other, not both. |
| `kmsConfig` | [KMS](#kms-configuration) | | Configuration of one or multiple KMS |
| `kmsConnectionName` | String | | Name of a [KMS external connection](#name-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](#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](#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](#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](#name-a-kms-external-connection) | `kmsConnectionName` | Once, in Gateway's environment | `keyId` |
| [Configure a KMS inline](#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](#kms-integration).
#### 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](/guide/conduktor-in-production/deploy-artifacts/deploy-gateway/external-connections#connect-to-a-kms).
Instead of a `kmsConfig` block, an encryption or decryption Interceptor can name a [KMS external connection](/guide/conduktor-in-production/deploy-artifacts/deploy-gateway/external-connections#connect-to-a-kms) 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 `` of a KMS connection, declared with `GATEWAY_EXTERNAL_CONNECTION__CONFIG_` 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://` | `in-memory-kms://my-password-key-id` |
| Azure | azure-kms\:// | `azure-kms://://.vault.azure.net/keys//` | `azure-kms://https://my-key-vault.vault.azure.net/keys/conduktor-gateway/4ceb7a4d1f3e4738b23bea870ae8745d` |
| AWS | aws-kms\:// | `aws-kms://arn:aws:kms:::key/` | `aws-kms://arn:aws:kms:us-east-1:123456789012:key/password-key-id` |
| GCP | gcp-kms\:// | `gcp-kms://projects//locations//keyRings//cryptoKeys/` | `gcp-kms://projects/my-project/locations/us-east1/keyRings/my-key-ring/cryptoKeys/password-key-id` |
| Fortanix | fortanix-kms\:// | `fortanix-kms://` | `fortanix-kms://password-key-id` |
| Gateway | gateway-kms\:// | `gateway-kms://` (uses master key from KMS above) | `gateway-kms://user-{{record.key}}` |
| Vault (Transit) | vault-kms\:// | `vault-kms://[scheme://]/transit/keys/` | `vault-kms://https://vault:8200/transit/keys/password-key-id` |
| Vault Transform | vault-transform:// | `vault-transform://` | `vault-transform://password-role` |
| Test tokenization | test-tokenization:// | `test-tokenization://` | `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](#in-memory-kms) | Local development and testing only | Keys don't persist - data becomes unreadable after Gateway restart |
| [Gateway](#gateway-kms) | Crypto shredding with high-volume per-record keys | Stores encrypted keys locally, reducing KMS costs for millions of unique keys |
| [AWS KMS](#aws-kms) | AWS infrastructure | Native IAM integration, no credential management needed |
| [Azure Key Vault](#azure-kms) | Azure infrastructure | Managed identity support, integrates with Azure services |
| [Vault KMS](#vault-kms) | Multi-cloud or on-premises | Most flexible authentication (11 methods), works anywhere |
| [GCP KMS](#google-cloud-platform-kms) | Google Cloud infrastructure | Service account integration, follows GCP security model |
| [Fortanix KMS](#fortanix-kms) | FIPS 140-2 compliance required | Hardware security module (HSM) backed, meets strict regulatory requirements |
#### 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
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](#kms-integration). 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:
```json theme={null}
"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:
```json theme={null}
"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:
```json theme={null}
{"algorithm":"AES128_GCM","keyId":"gateway-kms://fieldKeySecret-name-123456","uuid":""}
```
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:
```json theme={null}
{"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:
```json theme={null}
"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:
```json theme={null}
{"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
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](/guide/conduktor-in-production/deploy-artifacts/deploy-gateway/external-connections#aws-kms) carries the region and account, so each `keyId` is a bare key name rather than an ARN. Gateway builds the ARN for you:
```
arn::kms:::key/
```
* **The partition comes from the region**, so GovCloud, China and the ISO regions work without naming one.
* **An `alias/` id is used as the ARN resource as-is**, so it becomes `…::alias/` 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/'; the region and account are part of connection `. 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](#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**](https://docs.aws.amazon.com/sdk-for-java/v1/developer-guide/credentials.html#credentials-default) | | Configure the KMS from the context, not using variables. This will be overwritten if a specific KMS is configured within the Interceptor. |
#### 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](#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**](https://learn.microsoft.com/en-us/java/api/com.azure.identity.defaultazurecredential?view=azure-java-stable) | | 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
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](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
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](https://cloud.google.com/iam/docs/keys-create-delete#creating).
For enhanced security, you can hide the sensitive values using [environment variables as secrets](#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**](https://github.com/googleapis/google-auth-library-java/blob/main/README.md#application-default-credentials) | | 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
Gateway supports two data security backends for Vault:
* [Transit Secrets Engine](https://developer.hashicorp.com/vault/docs/secrets/transit) - the default and most useful
* [Transform Secrets Engine](https://developer.hashicorp.com/vault/docs/secrets/transform) - supports non-standard features like tokenization but requires [Vault Enterprise account](https://developer.hashicorp.com/vault/tutorials/get-started/available-editions#available-vault-editions))
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](#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:
```json theme={null}
{
"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:
```json theme={null}
{
"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:
```json theme={null}
{
"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:
```json theme={null}
{
"name": "myEncryptPlugin",
"pluginClass": "io.conduktor.gateway.interceptor.EncryptPlugin",
"config": {
"throttleTimeMs": 5000
}
}
```