> ## 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. # Kafka encryption configuration > Encrypt Kafka data with Conduktor Gateway: full payload, field-level and schema-based examples, with Conduktor CLI and API configurations. ## Overview This page gives you copy-paste examples to encrypt and decrypt Kafka data with Conduktor Gateway. Each example is an end-to-end scenario: a message arrives, an encryption Interceptor protects it on produce, and a `DecryptPlugin` returns it on consume. Find out more about: * [Encryption use cases](/guide/use-cases/kafka-encryption) * [Encryption reference](/guide/reference/data-security#encryption-interceptors) **From our blog:** [Stop building Kafka encryption libraries](https://conduktor.io/blog/stop-building-kafka-encryption-libraries) Why custom encryption code creates tech debt and how a proxy-based approach eliminates it. On-consume encryption plugins (`FetchEncryptPlugin` and `FetchEncryptSchemaBasedPlugin`) were removed in [Gateway v3.19.0](/guide/release-notes#gateway-3-19-0). To protect data on consume, use produce-side encryption or [data masking](/guide/conduktor-in-production/admin/data-masking). See [removed Interceptors](/guide/reference/removed-interceptors) for their previous documentation. ### How to read these examples The examples share one sample record, a customer with Personally Identifiable Information (PII): ```json theme={null} { "id": "cust-001", "email": "laura@example.com", "ssn": "123-45-6789", "phone": "+1-555-0100" } ``` * Field-level examples encrypt the `email`, `ssn` and `phone` fields of this record. * Apply a CLI example with `conduktor apply -f .yaml`. * The examples keep secrets (Vault tokens, Schema Registry credentials) out of the configuration by referencing environment variables set in the Gateway container. Use `$${VAR}` in the CLI examples and `${VAR}` in the curl examples: both resolve in Gateway. Find out more about [environment variables for secrets](/guide/conduktor-in-production/automate/cli-automation#environment-variables-for-secrets). ## Full payload encryption Use full payload encryption to protect the entire record value without looking inside it. Gateway treats the value as one blob, so you don't need a Schema Registry. Scenario: any message produced to a topic has its whole value encrypted on produce, and decrypted on consume. ### Encrypt the full payload ```yaml theme={null} apiVersion: gateway/v2 kind: Interceptor metadata: name: full-payload-encrypt spec: pluginClass: io.conduktor.gateway.interceptor.EncryptPlugin priority: 100 config: topic: ".*" kmsConfig: vault: uri: http://vault:8200 token: $${VAULT_TOKEN} recordValue: payload: keySecretId: vault-kms://vault:8200/transit/keys/payload-key algorithm: AES256_GCM ``` ```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": "full-payload-encrypt" }, "spec": { "pluginClass": "io.conduktor.gateway.interceptor.EncryptPlugin", "priority": 100, "config": { "topic": ".*", "kmsConfig": { "vault": { "uri": "http://vault:8200", "token": "${VAULT_TOKEN}" } }, "recordValue": { "payload": { "keySecretId": "vault-kms://vault:8200/transit/keys/payload-key", "algorithm": "AES256_GCM" } } } } }' ``` ### Decrypt the full payload On consume, a `DecryptPlugin` with access to the same KMS returns the original value. You don't list any fields, so Gateway decrypts the whole payload. ```yaml theme={null} apiVersion: gateway/v2 kind: Interceptor metadata: name: full-payload-decrypt spec: pluginClass: io.conduktor.gateway.interceptor.DecryptPlugin priority: 100 config: topic: ".*" kmsConfig: vault: uri: http://vault:8200 token: $${VAULT_TOKEN} ``` ```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": "full-payload-decrypt" }, "spec": { "pluginClass": "io.conduktor.gateway.interceptor.DecryptPlugin", "priority": 100, "config": { "topic": ".*", "kmsConfig": { "vault": { "uri": "http://vault:8200", "token": "${VAULT_TOKEN}" } } } } }' ``` ## Field-level encryption Use field-level encryption to protect named fields and leave the rest of the record readable. Each field can use its own key and algorithm, so you can grant different consumers access to different fields. Scenario: your producers send the sample customer record, and you want to encrypt `email`, `ssn` and `phone`, each with its own key. The field path syntax is the same whether the value is plain JSON or Avro. The one difference is that Avro records go through the Schema Registry, so the Avro example adds a `schemaRegistryConfig` block. Plain JSON needs no Schema Registry. ### Encrypt selected fields in a JSON message ```yaml theme={null} apiVersion: gateway/v2 kind: Interceptor metadata: name: field-encrypt-json spec: pluginClass: io.conduktor.gateway.interceptor.EncryptPlugin priority: 100 config: topic: ".*" kmsConfig: vault: uri: http://vault:8200 token: $${VAULT_TOKEN} recordValue: fields: - fieldName: email keySecretId: vault-kms://vault:8200/transit/keys/email-key algorithm: AES256_GCM - fieldName: ssn keySecretId: vault-kms://vault:8200/transit/keys/ssn-key algorithm: AES256_GCM - fieldName: phone keySecretId: vault-kms://vault:8200/transit/keys/phone-key algorithm: AES256_GCM ``` ```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": "field-encrypt-json" }, "spec": { "pluginClass": "io.conduktor.gateway.interceptor.EncryptPlugin", "priority": 100, "config": { "topic": ".*", "kmsConfig": { "vault": { "uri": "http://vault:8200", "token": "${VAULT_TOKEN}" } }, "recordValue": { "fields": [ { "fieldName": "email", "keySecretId": "vault-kms://vault:8200/transit/keys/email-key", "algorithm": "AES256_GCM" }, { "fieldName": "ssn", "keySecretId": "vault-kms://vault:8200/transit/keys/ssn-key", "algorithm": "AES256_GCM" }, { "fieldName": "phone", "keySecretId": "vault-kms://vault:8200/transit/keys/phone-key", "algorithm": "AES256_GCM" } ] } } } }' ``` ### Encrypt selected fields in an Avro message The config is identical, plus a `schemaRegistryConfig` block so Gateway can deserialize the Avro record and find the fields. This example also secures the Schema Registry with environment variables. ```yaml theme={null} apiVersion: gateway/v2 kind: Interceptor metadata: name: field-encrypt-avro spec: pluginClass: io.conduktor.gateway.interceptor.EncryptPlugin priority: 100 config: topic: ".*" schemaRegistryConfig: host: http://schema-registry:8081 additionalConfigs: schema.registry.url: $${SR_URL} basic.auth.credentials.source: USER_INFO basic.auth.user.info: $${SR_BASIC_AUTH_USER_INFO} kmsConfig: vault: uri: http://vault:8200 token: $${VAULT_TOKEN} recordValue: fields: - fieldName: email keySecretId: vault-kms://vault:8200/transit/keys/email-key algorithm: AES256_GCM - fieldName: ssn keySecretId: vault-kms://vault:8200/transit/keys/ssn-key algorithm: AES256_GCM - fieldName: phone keySecretId: vault-kms://vault:8200/transit/keys/phone-key algorithm: AES256_GCM ``` ```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": "field-encrypt-avro" }, "spec": { "pluginClass": "io.conduktor.gateway.interceptor.EncryptPlugin", "priority": 100, "config": { "topic": ".*", "schemaRegistryConfig": { "host": "http://schema-registry:8081", "additionalConfigs": { "schema.registry.url": "${SR_URL}", "basic.auth.credentials.source": "USER_INFO", "basic.auth.user.info": "${SR_BASIC_AUTH_USER_INFO}" } }, "kmsConfig": { "vault": { "uri": "http://vault:8200", "token": "${VAULT_TOKEN}" } }, "recordValue": { "fields": [ { "fieldName": "email", "keySecretId": "vault-kms://vault:8200/transit/keys/email-key", "algorithm": "AES256_GCM" }, { "fieldName": "ssn", "keySecretId": "vault-kms://vault:8200/transit/keys/ssn-key", "algorithm": "AES256_GCM" }, { "fieldName": "phone", "keySecretId": "vault-kms://vault:8200/transit/keys/phone-key", "algorithm": "AES256_GCM" } ] } } } }' ``` ### Decrypt all fields To return every encrypted field to the consumer, deploy a `DecryptPlugin` with no field list. Gateway decrypts all the fields it can, using the same KMS. ```yaml theme={null} apiVersion: gateway/v2 kind: Interceptor metadata: name: field-decrypt-all spec: pluginClass: io.conduktor.gateway.interceptor.DecryptPlugin priority: 100 config: topic: ".*" kmsConfig: vault: uri: http://vault:8200 token: $${VAULT_TOKEN} ``` ```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": "field-decrypt-all" }, "spec": { "pluginClass": "io.conduktor.gateway.interceptor.DecryptPlugin", "priority": 100, "config": { "topic": ".*", "kmsConfig": { "vault": { "uri": "http://vault:8200", "token": "${VAULT_TOKEN}" } } } } }' ``` To decrypt Avro, JSON Schema or Protobuf records, add a `schemaRegistryConfig` block to the `DecryptPlugin` config, as shown in the Avro encryption example above. ### Decrypt specific fields for a user To grant different consumers access to different fields, scope the `DecryptPlugin` and list only the fields that consumer may read with `recordValueFields`. Set the scope with `metadata.scope` (`username`, `group` or `vCluster`) in both the CLI and the API. Scenario: support agents (user `support-agent`) may read `phone` only, and the compliance team (group `compliance`) may read `email` and `ssn`. ```yaml theme={null} # Support agents can decrypt phone only apiVersion: gateway/v2 kind: Interceptor metadata: name: decrypt-support scope: username: support-agent spec: pluginClass: io.conduktor.gateway.interceptor.DecryptPlugin priority: 100 config: topic: ".*" kmsConfig: vault: uri: http://vault:8200 token: $${VAULT_TOKEN} recordValueFields: - phone --- # The compliance group can decrypt email and ssn apiVersion: gateway/v2 kind: Interceptor metadata: name: decrypt-compliance scope: group: compliance spec: pluginClass: io.conduktor.gateway.interceptor.DecryptPlugin priority: 100 config: topic: ".*" kmsConfig: vault: uri: http://vault:8200 token: $${VAULT_TOKEN} recordValueFields: - email - ssn ``` Create the support agent Interceptor, scoped by username in `metadata.scope`: ```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": "decrypt-support", "scope": { "username": "support-agent" } }, "spec": { "pluginClass": "io.conduktor.gateway.interceptor.DecryptPlugin", "priority": 100, "config": { "topic": ".*", "kmsConfig": { "vault": { "uri": "http://vault:8200", "token": "${VAULT_TOKEN}" } }, "recordValueFields": [ "phone" ] } } }' ``` Create the compliance Interceptor, scoped by group in `metadata.scope`: ```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": "decrypt-compliance", "scope": { "group": "compliance" } }, "spec": { "pluginClass": "io.conduktor.gateway.interceptor.DecryptPlugin", "priority": 100, "config": { "topic": ".*", "kmsConfig": { "vault": { "uri": "http://vault:8200", "token": "${VAULT_TOKEN}" } }, "recordValueFields": [ "email", "ssn" ] } } }' ``` A support agent reads the `phone` field in clear text while `email` and `ssn` stay encrypted. The compliance team reads `email` and `ssn` while `phone` stays encrypted. ## Schema-based field-level encryption Schema-based encryption is field-level encryption where you mark the fields to encrypt in the schema itself, instead of listing them in the Interceptor. Use it when you want the schema to own the encryption rules. It needs a Schema Registry and uses the `EncryptSchemaBasedPlugin`. ### Tag fields in your schema Add Conduktor constraints to the fields you want to encrypt, using the default `conduktor.` namespace. A field is encrypted when it has a `conduktor.keySecretId`, a `conduktor.algorithm`, or a `conduktor.tags` value that matches a tag in the Interceptor. ```json theme={null} { "$schema": "http://json-schema.org/draft-07/schema#", "title": "Customer", "type": "object", "properties": { "id": { "type": "string" }, "email": { "type": "string", "conduktor.keySecretId": "vault-kms://vault:8200/transit/keys/email-key", "conduktor.algorithm": "AES256_GCM" }, "ssn": { "type": "string", "conduktor.tags": ["PII"] }, "phone": { "type": "string", "conduktor.tags": ["PII"] } } } ``` ```json theme={null} { "type": "record", "name": "Customer", "fields": [ {"name": "id", "type": "string"}, { "name": "email", "type": "string", "conduktor.keySecretId": "vault-kms://vault:8200/transit/keys/email-key", "conduktor.algorithm": "AES256_GCM" }, {"name": "ssn", "type": "string", "conduktor.tags": ["PII"]}, {"name": "phone", "type": "string", "conduktor.tags": ["PII"]} ] } ``` Here, `email` uses the key and algorithm set in the schema. `ssn` and `phone` carry the `PII` tag, so they use the `defaultKeySecretId` and `defaultAlgorithm` from the Interceptor. For the Protobuf syntax, see the [encryption reference](/guide/reference/data-security#schema-based). ### Encrypt fields defined in the schema Deploy the `EncryptSchemaBasedPlugin` with the tags to look for and the defaults to apply. ```yaml theme={null} apiVersion: gateway/v2 kind: Interceptor metadata: name: schema-based-encrypt spec: pluginClass: io.conduktor.gateway.interceptor.EncryptSchemaBasedPlugin priority: 100 config: topic: ".*" schemaRegistryConfig: host: http://schema-registry:8081 additionalConfigs: schema.registry.url: $${SR_URL} basic.auth.credentials.source: USER_INFO basic.auth.user.info: $${SR_BASIC_AUTH_USER_INFO} kmsConfig: vault: uri: http://vault:8200 token: $${VAULT_TOKEN} defaultKeySecretId: vault-kms://vault:8200/transit/keys/pii-default-key defaultAlgorithm: AES256_GCM tags: - PII namespace: conduktor. ``` ```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": "schema-based-encrypt" }, "spec": { "pluginClass": "io.conduktor.gateway.interceptor.EncryptSchemaBasedPlugin", "priority": 100, "config": { "topic": ".*", "schemaRegistryConfig": { "host": "http://schema-registry:8081", "additionalConfigs": { "schema.registry.url": "${SR_URL}", "basic.auth.credentials.source": "USER_INFO", "basic.auth.user.info": "${SR_BASIC_AUTH_USER_INFO}" } }, "kmsConfig": { "vault": { "uri": "http://vault:8200", "token": "${VAULT_TOKEN}" } }, "defaultKeySecretId": "vault-kms://vault:8200/transit/keys/pii-default-key", "defaultAlgorithm": "AES256_GCM", "tags": [ "PII" ], "namespace": "conduktor." } } }' ``` ### Decrypt schema-based fields Decryption is the same for every field-level Interceptor, so you use the same `DecryptPlugin`. Because the records are Avro or JSON Schema, add a `schemaRegistryConfig` block. With no field list, Gateway decrypts every encrypted field. ```yaml theme={null} apiVersion: gateway/v2 kind: Interceptor metadata: name: schema-based-decrypt spec: pluginClass: io.conduktor.gateway.interceptor.DecryptPlugin priority: 100 config: topic: ".*" schemaRegistryConfig: host: http://schema-registry:8081 additionalConfigs: schema.registry.url: $${SR_URL} basic.auth.credentials.source: USER_INFO basic.auth.user.info: $${SR_BASIC_AUTH_USER_INFO} kmsConfig: vault: uri: http://vault:8200 token: $${VAULT_TOKEN} ``` ```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": "schema-based-decrypt" }, "spec": { "pluginClass": "io.conduktor.gateway.interceptor.DecryptPlugin", "priority": 100, "config": { "topic": ".*", "schemaRegistryConfig": { "host": "http://schema-registry:8081", "additionalConfigs": { "schema.registry.url": "${SR_URL}", "basic.auth.credentials.source": "USER_INFO", "basic.auth.user.info": "${SR_BASIC_AUTH_USER_INFO}" } }, "kmsConfig": { "vault": { "uri": "http://vault:8200", "token": "${VAULT_TOKEN}" } } } } }' ``` ## Combine Interceptors You can stack encryption Interceptors to apply different strategies to different parts of your data. A common pattern is field-level encryption for known sensitive fields, with full payload encryption as a fallback for messages that don't match. This relies on two settings: * **`priority`:** Interceptors run in priority order, lowest first. * **`errorPolicy: skip_already_encrypted`:** an encryption Interceptor skips records a previous Interceptor already encrypted. ### Encrypt fields with full payload fallback The field-level Interceptor runs first (priority 1) and encrypts `email`, `ssn` and `phone`. The full payload Interceptor runs second (priority 2) and encrypts anything left, skipping records that were already encrypted. ```yaml theme={null} # Step 1: field-level encryption (priority 1, runs first) apiVersion: gateway/v2 kind: Interceptor metadata: name: field-level-encrypt spec: pluginClass: io.conduktor.gateway.interceptor.EncryptPlugin priority: 1 config: topic: "sensitive-.*" kmsConfig: vault: uri: http://vault:8200 token: $${VAULT_TOKEN} recordValue: fields: - fieldName: email keySecretId: vault-kms://vault:8200/transit/keys/email-key algorithm: AES256_GCM - fieldName: ssn keySecretId: vault-kms://vault:8200/transit/keys/ssn-key algorithm: AES256_GCM - fieldName: phone keySecretId: vault-kms://vault:8200/transit/keys/phone-key algorithm: AES256_GCM --- # Step 2: full payload fallback (priority 2, runs second) apiVersion: gateway/v2 kind: Interceptor metadata: name: full-payload-fallback spec: pluginClass: io.conduktor.gateway.interceptor.EncryptPlugin priority: 2 config: topic: "sensitive-.*" errorPolicy: skip_already_encrypted kmsConfig: vault: uri: http://vault:8200 token: $${VAULT_TOKEN} recordValue: payload: keySecretId: vault-kms://vault:8200/transit/keys/payload-key algorithm: AES256_GCM ``` Step 1: create the field-level Interceptor (priority 1, runs first): ```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": "field-level-encrypt" }, "spec": { "pluginClass": "io.conduktor.gateway.interceptor.EncryptPlugin", "priority": 1, "config": { "topic": "sensitive-.*", "kmsConfig": { "vault": { "uri": "http://vault:8200", "token": "${VAULT_TOKEN}" } }, "recordValue": { "fields": [ { "fieldName": "email", "keySecretId": "vault-kms://vault:8200/transit/keys/email-key", "algorithm": "AES256_GCM" }, { "fieldName": "ssn", "keySecretId": "vault-kms://vault:8200/transit/keys/ssn-key", "algorithm": "AES256_GCM" }, { "fieldName": "phone", "keySecretId": "vault-kms://vault:8200/transit/keys/phone-key", "algorithm": "AES256_GCM" } ] } } } }' ``` Step 2: create the full payload fallback (priority 2, runs second): ```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": "full-payload-fallback" }, "spec": { "pluginClass": "io.conduktor.gateway.interceptor.EncryptPlugin", "priority": 2, "config": { "topic": "sensitive-.*", "errorPolicy": "skip_already_encrypted", "kmsConfig": { "vault": { "uri": "http://vault:8200", "token": "${VAULT_TOKEN}" } }, "recordValue": { "payload": { "keySecretId": "vault-kms://vault:8200/transit/keys/payload-key", "algorithm": "AES256_GCM" } } } } }' ``` **How it works:** 1. A message arrives on a `sensitive-.*` topic. 2. The field-level Interceptor (priority 1) runs first. If the message has `email`, `ssn` or `phone`, it encrypts those fields and flags the record. 3. The full payload Interceptor (priority 2) runs second. Because `errorPolicy` is `skip_already_encrypted`, it skips records the field-level Interceptor already handled. Records with no matching fields get full payload encryption instead. 4. On consume, a single `DecryptPlugin` handles both encryption types. Find out more about the [encryption error policy](/guide/reference/data-security#encryption-error-policy) in the reference documentation. If you also offload large messages to cloud storage, see [Combine with encryption](/guide/use-cases/manage-large-messages#combine-with-encryption) for the recommended priority order so ciphertext, not plaintext, lands in your cloud storage. ## Related resources * [Find out more about Interceptors](/guide/conduktor-concepts/interceptors) * [Encryption reference](/guide/reference/data-security#encryption-interceptors) * [Create data masking policies in Console](/guide/conduktor-in-production/admin/data-masking) * [Give us feedback/request a feature](https://conduktor.io/roadmap)