> ## 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) |
**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.
## 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-preview) 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 '{
"name": "myAuditInterceptorPlugin",
"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. |
#### 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.
### Schema registry
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` | `true` / `false` flag to determine whether the credentials provided should be validated when set. |
| `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/).
### 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 '{
"name": "myFieldLevelDataMaskingPlugin",
"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 '{
"name": "myFieldLevelDataMaskingPlugin",
"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) to the messages as they are produced through Gateway. We support templating in this format: `X-CLIENT_IP: "{{userIp}} testing"`.
### Context variables
These values are available as template variables:
* uuid
* userIp
* vcluster
* user
* clientId
* gatewayIp
* gatewayHost
* gatewayVersion
* apiKey
* apiKeyVersion
* timestampMillis
#### 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, you have to configure `schemaRegistryConfig` so Gateway can deserialize the records. For plain JSON data, no schema registry is needed.
### Configure header injection Interceptor
| Config | Type | Default | Description |
| :------------------- | :---------------------------------- | :------ | :-------------------------------------------------------------------------------------------------------------------------------- |
| topic | String | | Regular expression that matches topics from your produce request. |
| 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 fail the request if header injection fails. When `false`, errors are logged and the request continues. |
| schemaRegistryConfig | [Schema registry](#schema-registry) | | Schema Registry configuration. Required when using `record.key.field` or `record.value.field` templates with schema-encoded data. |
#### Error handling
The `failOnError` setting controls how the Interceptor handles errors during header injection:
* When `false` (default): errors are logged as warnings and the request continues processing. Headers that fail to be injected are skipped.
* When `true`: any error during header injection (such as deserialization failures or missing fields) causes the request to fail with a clear error message.
### 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 '{
"name": "myDynamicHeaderInjectionInterceptor",
"pluginClass": "io.conduktor.gateway.interceptor.DynamicHeaderInjectionPlugin",
"priority": 100,
"config": {
"topic": "topic.*",
"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.*
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. |
| `schemaDataMode` | String | `preserve_avro` | As of Gateway v3.3, you can decide to preserve the inbound message format when it encrypts the data **if** the incoming data is Avro, rather than converting the message to JSON (as per current behavior).
To convert the record to JSON and break the link to its schema in the backing topic, you can set this field to `convert_json` (default until v3.3). |
| `kmsConfig` | [KMS](#kms-configuration) | | Configuration of one or multiple KMS. |
| `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` | [Secret key template](#secret-keys) | | Default `keySecretId` to use if none is set in the schema. It must be a unique identifier for the secret key, and can be a template for Crypto Shredding use cases. |
| `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` | [Secret key template](#secret-keys) | | Secret key, can be a template for Crypto Shredding use cases. |
| `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` | [Secret Key template](#secret-keys) | `AES128_GCM` | Unique identifier for the secret key. You can store this key in your KMS by using the KMS key templates. It can be a template for crypto shredding use cases. |
| `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` | [Secret key template](#secret-keys) | | Secret key, can be a template for crypto shredding use cases. |
| `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` | [Secret key template](#secret-keys) | `AES128_GCM` | Unique identifier for the secret key. It can be a template for crypto shredding use cases. |
| `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` | [Secret key template](#secret-keys) | | Unique identifier for the secret key, and 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 `keySecretId` and `algorithm` specified in the schema.
If these are not defined in the schema, the `defaultKeySecretId` 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.
[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 `keySecretId`. That way, your secret keys will be 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" }]
}
```
You can set `"keySecretId": "{{record.topic}}-{{record.header.someHeader}}-{{record.key}}"` - this will create an encryption key called `myTopic-myHeader-myKey` in memory.
If you want this key to be stored in your Vault KMS, you can set: `"keySecretId": "vault-kms://https://vault:8200/transit/keys/{{record.topic}}-{{record.header.someHeader}}-{{record.key}}"`.
#### KMS integration
Any `keySecretId` 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.
If you want to make sure the key is well created in your KMS, you have to (1) [make sure you have configured the connection to the KMS](/guide/reference/data-security#kms-configuration) and (2) use the following format as `keySecretId`:
| 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` |
| 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` |
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.
#### Tokenization
Tokenization is an alternative to encryption for protecting sensitive data in Kafka messages. Instead of encrypting values, sensitive data is replaced with tokens, while also storing the original values securely in HashiCorp Vault's transform secrets engine.
Key benefits of tokenization:
* **Deterministic tokens**: the same input always generates the same token, enabling queries on tokenized data
* **Format-preserving transformation**: tokens can maintain the format of the original data
Tokenization is configured using the existing encryption Interceptors (`EncryptPlugin`, `DecryptPlugin`, etc.) but only works with the [Transform Secrets Engine](https://developer.hashicorp.com/vault/docs/secrets/transform) in Vault Enterprise ([find out how to configure Vault](#vault-kms)).
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.
Here's a sample field configured for tokenization:
```yaml theme={null}
fields:
- fieldName: "email"
keySecretId: "vault-transform://email-key"
```
#### 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}"
}
}
}
}
```
### 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. |
| `kmsConfig` | [KMS](#kms-configuration) | | Configuration of one or multiple KMS |
| `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). |
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` | `true` / `false` flag to determine whether the credentials provided should be validated when set. |
| `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.
### KMS configuration
Find out how to configure the different KMS within your encrypt and decrypt Interceptors.
#### 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. |
#### 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, loosing 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.
For enhanced security, you can hide the sensitive values using [environment variables as secrets](#use-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](#use-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](#use-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](#use-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
}
}
```