Kafka encryption configuration

Overview

This page provides configuration examples for encryption/decryption Interceptor use cases. Find out more about:

Encryption examples

Schema based encryption examples

Fields containing specific information with (keySecretId, algorithm, tags match) will be encrypted.
Field would be encrypted with the associated keySecretId, algorithm, if any missed, would be encrypted with the associated default ones in the Interceptor configuration.
Field would be encrypted with defaultSecret, defaultAlgorithm when tags has element with is in the Interceptor configuration.

Example for json schema, assuming we’re using default namespace (conduktor.):

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "Customer",
  "type": "object",
  "properties": {
    "name": { "type": "string" },
    "username": { "type": "string" },
    "password": { "type": "string", "conduktor.keySecretId": "vault-kms://vault:8200/transit/keys/password-secret", "conduktor.algorithm": "AES128_GCM"},
    "visa": { "type": "string", "conduktor.keySecretId": "vault-kms://vault:8200/transit/keys/password-visa"},
    "address": {
      "type": "object",
      "properties": {
        "location": { "type": "string", "conduktor.tags": ["MY_TAG", "PII", "GDPR", "MY_OTHER_TAG"]},
        "town": { "type": "string" },
        "country": { "type": "string" }
      }
    }
  }
}

Explanation:

password would be encrypted with the associated keySecretId, algorithm etc.
visa would be encryption with the associated keySecretId and the default algorithm provided in the Interceptor configuration.
location would be encrypted with defaultSecret, defaultAlgorithm because tags has PII with is in the Interceptor configuration.
fields containing no specific information (keySecretId, algorithm, tags without match) are left untouched.

Example for Avro schema, assuming we’re using default namespace (conduktor.):

{
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "name", "type": "string", "conduktor.algorithm": "AES128_GCM"},
    {"name": "age", "type": "int", "conduktor.keySecretId": "vault-kms://vault:8200/transit/keys/age-secret"},
    {"name": "email", "type": "string"},
    {
      "name": "address",
      "type": {
        "type": "record",
        "name": "AddressRecord",
        "fields": [
          {"name": "street", "type": "string", "conduktor.keySecretId": "vault-kms://vault:8200/transit/keys/street-secret"},
          {"name": "city", "type": "string", "conduktor.keySecretId": "vault-kms://vault:8200/transit/keys/city-secret", "conduktor.algorithm": "AES128_GCM"}
        ]
      }
    },
    {"name": "hobbies", "type": {"type": "array", "items": "string"}},
    {
      "name": "friends",
      "type": {
        "type": "array",
        "items": {
          "type": "record",
          "name": "Friend",
          "fields": [
            {"name": "name", "type": "string", "conduktor.tags": ["MY_TAG", "PII", "GDPR", "MY_OTHER_TAG"]},
            {"name": "age", "type": "int"}
          ]
        }
      }
    }
  ]
}

Example for Protobuf schema, assuming we’re using default namespace (conduktor.): In Protobuf, since we are using the Confluent schema registry, we use the (confluent.field_meta).params (with type map<string, string) for field options. Here’s how it can be defined:

  syntax = "proto3";
                 
  option java_package = "schema.protobuf";
  option java_outer_classname = "User";
  
  message Student {
    string name = 1 [(confluent.field_meta).params = {conduktor.keySecretId: "vault-kms://vault:8200/transit/keys/name-secret", conduktor.algorithm: "AES128_GCM"}];
    int32 age = 2 [(confluent.field_meta).params = {conduktor.keySecretId: "vault-kms://vault:8200/transit/keys/age-secret"}];
    string email = 3 [(confluent.field_meta).params = {conduktor.keySecretId: "vault-kms://vault:8200/transit/keys/email-secret"}];
    Address address = 4;
    repeated string hobbies = 5;
    repeated Friend friends = 6;
  
    message Address {
      string street = 1 [(confluent.field_meta).params = {conduktor.keySecretId: "vault-kms://vault:8200/transit/keys/street-secret", conduktor.algorithm: "AES128_GCM"}];
      string city = 2 [(confluent.field_meta).params = {conduktor.keySecretId: "vault-kms://vault:8200/transit/keys/city-secret"}];
    }
  
    message Friend {
      string name = 1 [(confluent.field_meta).params = {conduktor.tags: "[\"PII\", \"MY_TAG\"]"}];
      int32 age = 2 [(confluent.field_meta).params = {conduktor.keySecretId: "vault-kms://vault:8200/transit/keys/friend-age-secret"}];
    }
  }

Simple encrypt on produce

curl
Conduktor CLI

curl \
  --request PUT \
  --url 'http://localhost:8888/gateway/v2/interceptor' \
  --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
  --header 'Content-Type: application/json' \
  --data-raw '{
  "name": "myEncryptionPlugin",
  "pluginClass": "io.conduktor.gateway.interceptor.EncryptPlugin",
  "priority": 100,
  "config": {
    "topic": ".*",
    "recordValue": {
      "fields": [
        {
          "fieldName": "password",
          "keySecretId": "vault-kms://vault:8200/transit/keys/password-secret",
          "algorithm": "AES128_GCM"
        }
      ]
    }
  }
}'

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: myEncryptionPlugin
  scope:
    vCluster: passthrough
spec:
  pluginClass: io.conduktor.gateway.interceptor.EncryptPlugin
  priority: 100
  config:
    topic: ".*"
    recordValue:
      fields:
        - fieldName: password
          keySecretId: vault-kms://vault:8200/transit/keys/password-secret
          algorithm: AES128_GCM

Apply with:

conduktor apply -f simple-encrypt-on-produce.yaml

Field level encryption on produce

curl
Conduktor CLI

curl \
  --request PUT \
  --url 'http://localhost:8888/gateway/v2/interceptor' \
  --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
  --header 'Content-Type: application/json' \
  --data-raw '{
  "name": "myEncryptPlugin",
  "pluginClass": "io.conduktor.gateway.interceptor.EncryptPlugin",
  "priority": 100,
  "config": {
    "topic": ".*",
    "schemaRegistryConfig": {
      "host": "http://schema-registry:8081"
    },
    "kmsConfig": {
      "vault": {
        "uri": "https://vault:8200",
        "token": "vault-plaintext-root-token",
        "trustStore": {
          "trustStorePath": "/security/truststore.jks"
        }
      },
      "azure": {
        "tokenCredential": {
          "clientId": "azure_client_id",
          "tenantId": "azure_tenant_id",
          "clientSecret": "azure_client_secret"
        }
      }
    },
    "recordValue": {
      "fields": [
        {
          "fieldName": "password",
          "keySecretId": "vault-kms://vault:8200/transit/keys/password-secret",
          "algorithm": "AES128_GCM"
        },
        {
          "fieldName": "visa",
          "keySecretId": "vault-kms://vault:8200/transit/keys/{{record.header.test-header}}-visa-secret-{{record.key}}-{{record.value.username}}-{{record.value.education.account.accountId}}",
          "algorithm": "AES128_GCM"
        },
        {
          "fieldName": "education.account.username",
          "keySecretId": "azure-kms://https://my-key-vault.vault.azure.net/keys/conduktor-gateway/4ceb7a4d1f3e4738b23bea870ae8745d",
          "algorithm": "AES128_GCM"
        }
      ]
    }
  }
}'

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: myEncryptPlugin
  scope:
    vCluster: passthrough
spec:
  pluginClass: io.conduktor.gateway.interceptor.EncryptPlugin
  priority: 100
  config:
    topic: ".*"
    schemaRegistryConfig:
      host: http://schema-registry:8081
    kmsConfig:
      vault:
        uri: http://vault:8200
        token: vault-plaintext-root-token
        version: 1
      azure:
        tokenCredential:
          clientId: azure_client_id
          tenantId: azure_tenant_id
          clientSecret: azure_client_secret
    recordValue:
      fields:
        - fieldName: password
          keySecretId: vault-kms://vault:8200/transit/keys/password-secret
          algorithm: AES128_GCM
        - fieldName: visa
          keySecretId: vault-kms://vault:8200/transit/keys/{{record.header.test-header}}-visa-secret-{{record.key}}-{{record.value.username}}-{{record.value.education.account.accountId}}
          algorithm: AES128_GCM
        - fieldName: education.account.username
          keySecretId: azure-kms://https://my-key-vault.vault.azure.net/keys/conduktor-gateway/4ceb7a4d1f3e4738b23bea870ae8745d
          algorithm: AES128_GCM

Apply with:

conduktor apply -f field-level-encrypt-on-produce.yaml

Field-level encryption on produce with secured template

curl
Conduktor CLI

curl \
  --request PUT \
  --url 'http://localhost:8888/gateway/v2/interceptor' \
  --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
  --header 'Content-Type: application/json' \
  --data-raw '{
  "name": "myEncryptPlugin",
  "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": "${SR_BASIC_AUTH_CRED_SRC}",
        "basic.auth.user.info": "${SR_BASIC_AUTH_USER_INFO}"
      }
    },
    "kmsConfig": {
      "vault": {
        "uri": "https://vault:8200",
        "token": "${VAULT_TOKEN}",
        "version": 1
        "trustStore": {
          "trustStorePath": "/security/truststore.jks"
        }
      }
    },
    "recordValue": {
      "fields": [
        {
          "fieldName": "password",
          "keySecretId": "vault-kms://vault:8200/transit/keys/password-secret",
          "algorithm": "AES128_GCM"
        },
        {
          "fieldName": "visa",
          "keySecretId": "vault-kms://vault:8200/transit/keys/{{record.header.test-header}}-visa-secret-{{record.key}}-{{record.value.username}}-{{record.value.education.account.accountId}}",
          "algorithm": "AES128_GCM"
        },
        {
          "fieldName": "education.account.username",
          "keySecretId": "vault-kms://vault:8200/transit/keys/{{record.header.test-header}}-secret-key-account-username-{{record.topic}}",
          "algorithm": "AES128_GCM"
        }
      ]
    }
  }
}'

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: myEncryptPlugin
  scope:
    vCluster: passthrough
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: $${SR_BASIC_AUTH_CRED_SRC}
        basic.auth.user.info: $${SR_BASIC_AUTH_USER_INFO}
    kmsConfig:
      vault:
        uri: http://vault:8200
        token: $${VAULT_TOKEN}
        version: 1
    recordValue:
      fields:
        - fieldName: password
          keySecretId: vault-kms://vault:8200/transit/keys/password-secret
          algorithm: AES128_GCM
        - fieldName: visa
          keySecretId: vault-kms://vault:8200/transit/keys/{{record.header.test-header}}-visa-secret-{{record.key}}-{{record.value.username}}-{{record.value.education.account.accountId}}
          algorithm: AES128_GCM
        - fieldName: education.account.username
          keySecretId: vault-kms://vault:8200/transit/keys/{{record.header.test-header}}-secret-key-account-username-{{record.topic}}
          algorithm: AES128_GCM

Apply with:

conduktor apply -f field-level-encrypt-secured-template.yaml

Schema-based field level encryption on produce

curl
Conduktor CLI

curl \
  --request PUT \
  --url 'http://localhost:8888/gateway/v2/interceptor' \
  --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
  --header 'Content-Type: application/json' \
  --data-raw '{
  "name": "mySchemaBasedEncryptPlugin",
  "pluginClass": "io.conduktor.gateway.interceptor.EncryptSchemaBasedPlugin",
  "priority": 100,
  "config": {
    "topic": ".*",
    "schemaRegistryConfig": {
      "host": "http://schema-registry:8081"
    },
    "kmsConfig": {
      "vault": {
        "uri": "https://vault:8200",
        "token": "vault-plaintext-root-token",
        "version": 1
        "trustStore": {
          "trustStorePath": "/security/truststore.jks"
        }
      }
    },
    "defaultKeySecretId": "vault-kms://vault:8200/transit/keys/myDefaultKeySecret",
    "defaultAlgorithm": "AES128_EAX",
    "tags": ["PII", "ENCRYPTION"],
    "namespace": "conduktor."
  }
}'

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: mySchemaBasedEncryptPlugin
  scope:
    vCluster: passthrough
spec:
  pluginClass: io.conduktor.gateway.interceptor.EncryptSchemaBasedPlugin
  priority: 100
  config:
    topic: ".*"
    schemaRegistryConfig:
      host: http://schema-registry:8081
    kmsConfig:
      vault:
        uri: http://vault:8200
        token: vault-plaintext-root-token
        version: 1
    defaultKeySecretId: vault-kms://vault:8200/transit/keys/myDefaultKeySecret
    defaultAlgorithm: AES128_EAX
    tags:
      - PII
      - ENCRYPTION
    namespace: conduktor.

Apply with:

conduktor apply -f schema-based-encrypt-on-produce.yaml

Full message level encryption on produce

curl
Conduktor CLI

curl \
  --request PUT \
  --url 'http://localhost:8888/gateway/v2/interceptor' \
  --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
  --header 'Content-Type: application/json' \
  --data-raw '{
  "name": "myEncryptPlugin",
  "pluginClass": "io.conduktor.gateway.interceptor.EncryptPlugin",
  "priority": 100,
  "config": {
    "topic": ".*",
    "schemaRegistryConfig": {
      "host": "http://schema-registry:8081"
    },
    "kmsConfig": {
      "vault": {
        "uri": "https://vault:8200",
        "token": "vault-plaintext-root-token",
        "trustStore": {
          "trustStorePath": "/security/truststore.jks"
        }
      }
    },
    "recordValue": {
      "payload": {
        "keySecretId": "vault-kms://vault:8200/transit/keys/{{record.header.test-header}}-secret-key-account-username-{{record.topic}}",
        "algorithm": "AES128_GCM"
      }
    }
  }
}'

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: myEncryptPlugin
  scope:
    vCluster: passthrough
spec:
  pluginClass: io.conduktor.gateway.interceptor.EncryptPlugin
  priority: 100
  config:
    topic: ".*"
    schemaRegistryConfig:
      host: http://schema-registry:8081
    kmsConfig:
      vault:
        uri: http://vault:8200
        token: vault-plaintext-root-token
        version: 1
    recordValue:
      payload:
        keySecretId: vault-kms://vault:8200/transit/keys/{{record.header.test-header}}-secret-key-account-username-{{record.topic}}
        algorithm: AES128_GCM

Apply with:

conduktor apply -f full-message-encrypt-on-produce.yaml

Full message level encryption on produce with secured template

curl
Conduktor CLI

curl \
  --request PUT \
  --url 'http://localhost:8888/gateway/v2/interceptor' \
  --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
  --header 'Content-Type: application/json' \
  --data-raw '{
  "name": "myEncryptPlugin",
  "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": "${SR_BASIC_AUTH_CRED_SRC}",
        "basic.auth.user.info": "${SR_BASIC_AUTH_USER_INFO}"
      }
    },
    "kmsConfig": {
      "vault": {
        "uri": "https://vault:8200",
        "token": "${VAULT_TOKEN}",
        "trustStore": {
          "trustStorePath": "/security/truststore.jks"
        }
      }
    },
    "recordValue": {
      "payload": {
        "keySecretId": "vault-kms://vault:8200/transit/keys/{{record.header.test-header}}-secret-key-account-username-{{record.topic}}",
        "algorithm": "AES128_GCM"
      }
    }
  }
}'

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: myEncryptPlugin
  scope:
    vCluster: passthrough
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: $${SR_BASIC_AUTH_CRED_SRC}
        basic.auth.user.info: $${SR_BASIC_AUTH_USER_INFO}
    kmsConfig:
      vault:
        uri: http://vault:8200
        token: $${VAULT_TOKEN}
        version: 1
    recordValue:
      payload:
        keySecretId: vault-kms://vault:8200/transit/keys/$${record.header.test-header}-secret-key-account-username-$${record.topic}
        algorithm: AES128_GCM

Apply with:

conduktor apply -f full-message-encrypt-secured-template.yaml

Encryption on consume

On consume encryption plugins were deprecated in Gateway v3.16.0 and will be removed in Gateway v3.19.0.

curl
Conduktor CLI

curl \
  --request PUT \
  --url 'http://localhost:8888/gateway/v2/interceptor' \
  --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
  --header 'Content-Type: application/json' \
  --data-raw '{
  "name": "myEncryptPlugin",
  "pluginClass": "io.conduktor.gateway.interceptor.FetchEncryptPlugin",
  "priority": 100,
  "config": {
    "topic": ".*",
    "schemaRegistryConfig": {
      "host": "http://schema-registry:8081"
    },
    "kmsConfig": {
      "vault": {
        "uri": "http://vault:8200",
        "token": "vault-plaintext-root-token",
        "version": 1
      }
    },
    "recordValue": {
      "fields": [
        {
          "fieldName": "password",
          "keySecretId": "vault-kms://vault:8200/transit/keys/password-secret",
          "algorithm": "AES128_GCM"
        },
        {
          "fieldName": "visa",
          "keySecretId": "vault-kms://vault:8200/transit/keys/{{record.header.test-header}}-visa-secret-{{record.key}}-{{record.value.username}}-{{record.value.education.account.accountId}}",
          "algorithm": "AES128_GCM"
        },
        {
          "fieldName": "education.account.username",
          "keySecretId": "vault-kms://vault:8200/transit/keys/{{record.header.test-header}}-secret-key-account-username-{{record.topic}}",
          "algorithm": "AES128_GCM"
        }
      ]
    }
  }
}'

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: myEncryptPlugin
  scope:
    vCluster: passthrough
spec:
  pluginClass: io.conduktor.gateway.interceptor.FetchEncryptPlugin
  priority: 100
  config:
    topic: ".*"
    schemaRegistryConfig:
      host: http://schema-registry:8081
    kmsConfig:
      vault:
        uri: http://vault:8200
        token: vault-plaintext-root-token
        version: 1
    recordValue:
      fields:
        - fieldName: password
          keySecretId: vault-kms://vault:8200/transit/keys/password-secret
          algorithm: AES128_GCM
        - fieldName: visa
          keySecretId: vault-kms://vault:8200/transit/keys/{{record.header.test-header}}-visa-secret-{{record.key}}-{{record.value.username}}-{{record.value.education.account.accountId}}
          algorithm: AES128_GCM
        - fieldName: education.account.username
          keySecretId: vault-kms://vault:8200/transit/keys/{{record.header.test-header}}-secret-key-account-username-{{record.topic}}
          algorithm: AES128_GCM

Apply with:

conduktor apply -f encrypt-on-consume.yaml

Schema-based field level encryption on consume

On consume encryption plugins were deprecated in Gateway v3.16.0 and will be removed in Gateway v3.19.0.

curl
Conduktor CLI

curl \
  --request PUT \
  --url 'http://localhost:8888/gateway/v2/interceptor' \
  --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
  --header 'Content-Type: application/json' \
  --data-raw '{
  "name": "mySchemaBasedEncryptPlugin",
  "pluginClass": "io.conduktor.gateway.interceptor.FetchEncryptSchemaBasedPlugin",
  "priority": 100,
  "config": {
    "topic": ".*",
    "schemaRegistryConfig": {
      "host": "http://schema-registry:8081"
    },
    "kmsConfig": {
      "vault": {
        "uri": "https://vault:8200",
        "token": "vault-plaintext-root-token",
        "trustStore": {
          "trustStorePath": "/security/truststore.jks"
        }
      }
    },
    "defaultKeySecretId": "vault-kms://vault:8200/transit/keys/myDefaultKeySecret",
    "defaultAlgorithm": "AES128_EAX",
    "tags": ["PII", "ENCRYPTION"],
    "namespace": "conduktor."
  }
}'

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: mySchemaBasedEncryptPlugin
  scope:
    vCluster: passthrough
spec:
  pluginClass: io.conduktor.gateway.interceptor.FetchEncryptSchemaBasedPlugin
  priority: 100
  config:
    topic: ".*"
    schemaRegistryConfig:
      host: http://schema-registry:8081
    kmsConfig:
      vault:
        uri: http://vault:8200
        token: vault-plaintext-root-token
        version: 1
    defaultKeySecretId: vault-kms://vault:8200/transit/keys/myDefaultKeySecret
    defaultAlgorithm: AES128_EAX
    tags:
      - PII
      - ENCRYPTION
    namespace: conduktor.

Apply with:

conduktor apply -f schema-based-encrypt-on-consume.yaml

Encryption on consume with secured template

On consume encryption plugins were deprecated in Gateway v3.16.0 and will be removed in Gateway v3.19.0.

curl
Conduktor CLI

curl \
  --request PUT \
  --url 'http://localhost:8888/gateway/v2/interceptor' \
  --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
  --header 'Content-Type: application/json' \
  --data-raw '{
  "name": "myEncryptPlugin",
  "pluginClass": "io.conduktor.gateway.interceptor.FetchEncryptPlugin",
  "priority": 100,
  "config": {
    "topic": ".*",
    "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}"
      }
    },
    "kmsConfig": {
      "vault": {
        "uri": "https://vault:8200",
        "token": "${VAULT_TOKEN}",
        "trustStore": {
          "trustStorePath": "/security/truststore.jks"
        }
      }
    },
    "recordValue": {
      "fields": [
        {
          "fieldName": "password",
          "keySecretId": "vault-kms://vault:8200/transit/keys/password-secret",
          "algorithm": "AES128_GCM"
        },
        {
          "fieldName": "visa",
          "keySecretId": "vault-kms://vault:8200/transit/keys/{{record.header.test-header}}-visa-secret-{{record.key}}-{{record.value.username}}-{{record.value.education.account.accountId}}",
          "algorithm": "AES128_GCM"
        },
        {
          "fieldName": "education.account.username",
          "keySecretId": "vault-kms://vault:8200/transit/keys/{{record.header.test-header}}-secret-key-account-username-{{record.topic}}",
          "algorithm": "AES128_GCM"
        }
      ]
    }
  }
}'

apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: myEncryptPlugin
  scope:
    vCluster: passthrough
spec:
  pluginClass: io.conduktor.gateway.interceptor.FetchEncryptPlugin
  priority: 100
  config:
    topic: ".*"
    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}
    kmsConfig:
      vault:
        uri: http://vault:8200
        token: $${VAULT_TOKEN}
        version: 1
    recordValue:
      fields:
        - fieldName: password
          keySecretId: vault-kms://vault:8200/transit/keys/password-secret
          algorithm: AES128_GCM
        - fieldName: visa
          keySecretId: vault-kms://vault:8200/transit/keys/{{record.header.test-header}}-visa-secret-{{record.key}}-{{record.value.username}}-{{record.value.education.account.accountId}}
          algorithm: AES128_GCM
        - fieldName: education.account.username
          keySecretId: vault-kms://vault:8200/transit/keys/{{record.header.test-header}}-secret-key-account-username-{{record.topic}}
          algorithm: AES128_GCM

Apply with:

conduktor apply -f encrypt-on-consume-secured-template.yaml

Decryption examples

Decrypt all fields

{
  "name": "myDecryptPlugin",
  "pluginClass": "io.conduktor.gateway.interceptor.DecryptPlugin",
  "priority": 100,
  "config": {
    "topic": ".*",
    "schemaRegistryConfig": {
      "host": "http://schema-registry:8081"
    },
    "kmsConfig": {
      "vault": {
        "uri": "https://vault:8200",
        "token": "vault-plaintext-root-token",
        "trustStore": {
          "trustStorePath": "/security/truststore.jks"
        }
      }
    }
  }
}

Decrypt all fields with secured template

{
  "name": "myDecryptPlugin",
  "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": "${SR_BASIC_AUTH_CRED_SRC}",
        "basic.auth.user.info": "${SR_BASIC_AUTH_USER_INFO}"
      }
    },
    "kmsConfig": {
      "vault": {
        "uri": "https://vault:8200",
        "token": "${VAULT_TOKEN}",
        "trustStore": {
          "trustStorePath": "/security/truststore.jks"
        }
      }
    }
  }
}

Decrypt specific fields

{
  "name": "myDecryptPlugin",
  "pluginClass": "io.conduktor.gateway.interceptor.DecryptPlugin",
  "priority": 100,
  "config": {
    "topic": ".*",
    "schemaRegistryConfig": {
      "host": "http://schema-registry:8081"
    },
    "kmsConfig": {
      "vault": {
        "uri": "https://vault:8200",
        "token": "vault-plaintext-root-token",
        "trustStore": {
          "trustStorePath": "/security/truststore.jks"
        }
      }
    },
    "recordValueFields": [
      "visa",
      "education.account.username"
    ],
    "recordKeyFields": [
      "bank.accountNo"
    ],
    "recordHeaderFields": [
      "account.username"
    ]
  }
}

Decrypt specific fields with secured template

{
  "name": "myDecryptPlugin",
  "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": "${SR_BASIC_AUTH_CRED_SRC}",
        "basic.auth.user.info": "${SR_BASIC_AUTH_USER_INFO}"
      }
    },
    "kmsConfig": {
      "vault": {
        "uri": "https://vault:8200",
        "token": "${VAULT_TOKEN}",
        "trustStore": {
          "trustStorePath": "/security/truststore.jks"
        }
      }
    },
    "recordValueFields": [
      "visa",
      "education.account.username"
    ],
    "recordKeyFields": [
      "bank.accountNo"
    ],
    "recordHeaderFields": [
      "account.username"
    ]
  }
}

Decrypt full message

{
  "name": "myDecryptPlugin",
  "pluginClass": "io.conduktor.gateway.interceptor.DecryptPlugin",
  "priority": 100,
  "config": {
    "topic": ".*",
    "kmsConfig": {
      "vault": {
        "uri": "https://vault:8200",
        "token": "vault-plaintext-root-token",
        "trustStore": {
          "trustStorePath": "/security/truststore.jks"
        }
      }
    }
  }
}

Decrypt full message with secured template

{
  "name": "myDecryptPlugin",
  "pluginClass": "io.conduktor.gateway.interceptor.DecryptPlugin",
  "priority": 100,
  "config": {
    "topic": ".*",
    "kmsConfig": {
      "vault": {
        "uri": "https://vault:8200",
        "token": "${VAULT_TOKEN}",
        "trustStore": {
          "trustStorePath": "/security/truststore.jks"
        }
      }
    }
  }
}

Stack encryption Interceptors

You can combine multiple encryption Interceptors to apply different encryption strategies to different parts of your data. A common pattern is to use field-level encryption for known sensitive fields, with full payload encryption as a fallback for any messages that don’t match the field-level rules. This is controlled by the errorPolicy setting. When set to skip_already_encrypted, an encryption Interceptor skips records that were already encrypted by a previous Interceptor in the chain. Interceptors execute in priority order (lowest number first). To stack them:

Set a lower priority (runs first) for your field-level encryption Interceptor
Set a higher priority (runs second) for your full payload encryption Interceptor
Set errorPolicy: skip_already_encrypted on the full payload Interceptor so it only encrypts records that weren’t already handled

Field-level encryption with full payload fallback

In this example, the password and visa fields are encrypted individually. Any message that doesn’t have field-level encryption applied (for example, a message without those fields) gets full payload encryption as a fallback.

Conduktor CLI
curl

# Step 1: Field-level encryption (priority 1, runs first)
apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: fieldLevelEncrypt
  scope:
    vCluster: passthrough
spec:
  pluginClass: io.conduktor.gateway.interceptor.EncryptPlugin
  priority: 1
  config:
    topic: "sensitive-.*"
    kmsConfig:
      vault:
        uri: http://vault:8200
        token: ${VAULT_TOKEN}
    recordValue:
      fields:
        - fieldName: password
          keySecretId: vault-kms://vault:8200/transit/keys/password-key
          algorithm: AES256_GCM
        - fieldName: visa
          keySecretId: vault-kms://vault:8200/transit/keys/visa-key
          algorithm: AES256_GCM
---
# Step 2: Full payload encryption fallback (priority 2, runs second)
apiVersion: gateway/v2
kind: Interceptor
metadata:
  name: fullPayloadEncryptFallback
  scope:
    vCluster: passthrough
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/full-payload-key
        algorithm: AES256_GCM

Apply with:

conduktor apply -f stacked-encryption.yaml

Step 1: Create the field-level encryption Interceptor (priority 1, runs first):

curl \
  --request PUT \
  --url 'http://localhost:8888/gateway/v2/interceptor' \
  --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
  --header 'Content-Type: application/json' \
  --data-raw '{
  "name": "fieldLevelEncrypt",
  "pluginClass": "io.conduktor.gateway.interceptor.EncryptPlugin",
  "priority": 1,
  "config": {
    "topic": "sensitive-.*",
    "kmsConfig": {
      "vault": {
        "uri": "http://vault:8200",
        "token": "${VAULT_TOKEN}"
      }
    },
    "recordValue": {
      "fields": [
        {
          "fieldName": "password",
          "keySecretId": "vault-kms://vault:8200/transit/keys/password-key",
          "algorithm": "AES256_GCM"
        },
        {
          "fieldName": "visa",
          "keySecretId": "vault-kms://vault:8200/transit/keys/visa-key",
          "algorithm": "AES256_GCM"
        }
      ]
    }
  }
}'

Step 2: Create the full payload encryption fallback (priority 2, runs second):

curl \
  --request PUT \
  --url 'http://localhost:8888/gateway/v2/interceptor' \
  --header 'Authorization: Basic YWRtaW46Y29uZHVrdG9y' \
  --header 'Content-Type: application/json' \
  --data-raw '{
  "name": "fullPayloadEncryptFallback",
  "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/full-payload-key",
        "algorithm": "AES256_GCM"
      }
    }
  }
}'

How it works:

A message arrives on a sensitive-.* topic
The field-level Interceptor (priority 1) runs first. If the message contains password or visa fields, those fields are encrypted and the record is flagged with encryption headers
The full payload Interceptor (priority 2) runs second. Because errorPolicy is set to skip_already_encrypted, it skips any record that was already encrypted by the field-level Interceptor. Records that weren’t field-level encrypted get full payload encryption
On consume, a single DecryptPlugin handles decryption for both encryption types

Find out more about the encryption error policy in the reference documentation.

Tutorials

Deploy

Use

Configure

Migrate

Overview

Encryption examples

Schema based encryption examples

Simple encrypt on produce

Field level encryption on produce

Field-level encryption on produce with secured template

Schema-based field level encryption on produce

Full message level encryption on produce

Full message level encryption on produce with secured template

Encryption on consume

Schema-based field level encryption on consume

Encryption on consume with secured template

Decryption examples

Decrypt all fields

Decrypt all fields with secured template

Decrypt specific fields

Decrypt specific fields with secured template

Decrypt full message

Decrypt full message with secured template

Stack encryption Interceptors

Field-level encryption with full payload fallback

Tutorials

Deploy

Use

Configure

Migrate

​Overview

​Encryption examples

​Schema based encryption examples

​Simple encrypt on produce

​Field level encryption on produce

​Field-level encryption on produce with secured template

​Schema-based field level encryption on produce

​Full message level encryption on produce

​Full message level encryption on produce with secured template

​Encryption on consume

​Schema-based field level encryption on consume

​Encryption on consume with secured template

​Decryption examples

​Decrypt all fields

​Decrypt all fields with secured template

​Decrypt specific fields

​Decrypt specific fields with secured template

​Decrypt full message

​Decrypt full message with secured template

​Stack encryption Interceptors

​Field-level encryption with full payload fallback

​Related resources

Overview

Encryption examples

Schema based encryption examples

Simple encrypt on produce

Field level encryption on produce

Field-level encryption on produce with secured template

Schema-based field level encryption on produce

Full message level encryption on produce

Full message level encryption on produce with secured template

Encryption on consume

Schema-based field level encryption on consume

Encryption on consume with secured template

Decryption examples

Decrypt all fields

Decrypt all fields with secured template

Decrypt specific fields

Decrypt specific fields with secured template

Decrypt full message

Decrypt full message with secured template

Stack encryption Interceptors

Field-level encryption with full payload fallback

Related resources