Skip to main content

​
Overview

This page provides configuration examples for encryption/decryption Interceptor use cases. Find out more about:
From our blog: Stop building Kafka encryption libraries Why custom encryption code creates tech debt and how a proxy-based approach eliminates it.

​
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 \
  --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"
        }
      ]
    }
  }
}'

​
Field level encryption on produce

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"
        }
      ]
    }
  }
}'

​
Field-level encryption on produce with secured template

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"
        }
      ]
    }
  }
}'

​
Schema-based field level encryption on produce

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."
  }
}'

​
Full message level encryption on produce

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"
      }
    }
  }
}'

​
Full message level encryption on produce with secured template

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"
      }
    }
  }
}'
On consume encryption plugins (FetchEncryptPlugin and FetchEncryptSchemaBasedPlugin) were removed in Gateway v3.19.0. See removed Interceptors for their documentation.

​
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:
  1. Set a lower priority (runs first) for your field-level encryption Interceptor
  2. Set a higher priority (runs second) for your full payload encryption Interceptor
  3. 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. 
# 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
How it works:
  1. A message arrives on a sensitive-.* topic
  2. 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
  3. 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
  4. On consume, a single DecryptPlugin handles decryption for both encryption types
Find out more about the encryption error policy in the reference documentation.
If you also offload large messages to cloud storage, see Combine with encryption for the recommended priority order so ciphertext, not plaintext, lands in your cloud storage.