Skip to main content
Quick navigation

Schema based field level encryption with Schema Registry

Yes, it work with Avro, Json Schema with nested fields

View the full demo in realtime​

You can either follow all the steps manually, or watch the recording

Review the docker compose environment​

As can be seen from docker-compose.yaml the demo environment consists of the following services:

  • gateway1
  • gateway2
  • kafka-client
  • kafka1
  • kafka2
  • kafka3
  • schema-registry
  • vault
cat docker-compose.yaml

Starting the docker environment​

Start all your docker processes, wait for them to be up and ready, then run in background

  • --wait: Wait for services to be running|healthy. Implies detached mode.
  • --detach: Detached mode: Run containers in the background
docker compose up --detach --wait

Creating topic customers on gateway1​

Creating on gateway1:

  • Topic customers with partitions:1 and replication-factor:1
kafka-topics \
    --bootstrap-server localhost:6969 \
    --replication-factor 1 \
    --partitions 1 \
    --create --if-not-exists \
    --topic customers

Adding interceptor encrypt​

We want to encrypt two fields at the root layer, and location in the address object. Here we are using an in memory KMS.

step-06-encrypt-interceptor.json:

{
  "kind" : "Interceptor",
  "apiVersion" : "gateway/v2",
  "metadata" : {
    "name" : "encrypt"
  },
  "spec" : {
    "comment" : "Adding interceptor: encrypt",
    "pluginClass" : "io.conduktor.gateway.interceptor.EncryptSchemaBasedPlugin",
    "priority" : 100,
    "config" : {
      "schemaDataMode" : "convert_json",
      "kmsConfig" : {
        "vault" : {
          "uri" : "http://vault:8200",
          "token" : "vault-plaintext-root-token",
          "version" : 1
        }
      },
      "schemaRegistryConfig" : {
        "host" : "http://schema-registry:8081"
      },
      "defaultKeySecretId" : "myDefaultKeySecret",
      "defaultAlgorithm" : "AES128_EAX",
      "tags" : [ "PII", "ENCRYPTION" ],
      "namespace" : "conduktor."
    }
  }
}
curl \
    --silent \
    --request PUT "http://localhost:8888/gateway/v2/interceptor" \
    --header "Content-Type: application/json" \
    --user "admin:conduktor" \
    --data @step-06-encrypt-interceptor.json | jq

Listing interceptors​

Listing interceptors on gateway1

curl \
    --silent \
    --request GET "http://localhost:8888/gateway/v2/interceptor" \
    --user "admin:conduktor" | jq

Let's send unencrypted json schema message with specified json schema with custom constrains for encryption​

valueSchema=$(echo '{
    "title": "Customer",
    "type": "object",
    "properties": {
      "name": { "type": "string" },
      "username": { "type": "string" },
      "password": { "type": "string", "conduktor.keySecretId": "password-secret", "conduktor.algorithm": "AES128_GCM" },
      "visa": { "type": "string", "conduktor.keySecretId": "conduktor.visa-secret", "conduktor.algorithm": "AES128_GCM" },
      "address": {
        "type": "object",
        "properties": {
          "location": { "type": "string", "conduktor.tags": ["MY_TAG", "PII", "GDPR", "MY_OTHER_TAG"] },
          "town": { "type": "string" },
          "country": { "type": "string" }
        }
      }
    }
}' | jq -c)

keySchema=$(echo '{
    "title": "Metadata",
    "type": "object",
    "properties": {
        "sessionId": {"type": "string"},
        "authenticationToken": {"type": "string", "conduktor.keySecretId": "token-secret"},
        "deviceInformation": {"type": "string", "conduktor.algorithm": "AES128_CTR_HMAC_SHA256" }
    }
}' | jq -c)

invalidKeyTom=$(echo '{
        "sessionId": "session-id-tom",
        "authenticationToken": "authentication-token-tom",
        "deviceInformation": "device-information-tom"
    }' | jq -c)

invalidValueTom=$(echo '{
        "name": "tom",
        "username": "tom@conduktor.io",
        "password": "motorhead",
        "visa": "#abc123",
        "address": {
          "location": "12 Chancery lane",
          "town": "London",
          "country": "UK"
        }
    }' | jq -c)

invalidInputTom="$invalidKeyTom|$invalidValueTom"
echo $invalidInputTom | \
kafka-json-schema-console-producer \
        --bootstrap-server localhost:6969 \
        --topic customers \
        --property schema.registry.url=http://localhost:8081 \
        --property parse.key=true \
        --property key.separator="|" \
        --property value.schema=$valueSchema \
        --property key.schema=$keySchema 2>&1 /dev/null

invalidKeyLaura=$(echo '{
        "sessionId": "session-id-laura",
        "authenticationToken": "authentication-token-laura",
        "deviceInformation": "device-information-laura"
    }' | jq -c)

invalidValueLaura=$(echo '{
        "name": "laura",
        "username": "laura@conduktor.io",
        "password": "kitesurf",
        "visa": "#888999XZ;",
        "address": {
          "location": "4th Street, Jumeirah",
          "town": "Dubai",
          "country": "UAE"
        }
    }' | jq -c)

invalidInputLaura="$invalidKeyLaura|$invalidValueLaura"
echo $invalidInputLaura | \
kafka-json-schema-console-producer \
        --bootstrap-server localhost:6969 \
        --topic customers \
        --property schema.registry.url=http://localhost:8081 \
        --property parse.key=true \
        --property key.separator="|" \
        --property value.schema=$valueSchema \
        --property key.schema=$keySchema 2>&1 /dev/null

Let's make sure they are encrypted​

password and visa and the nested field address.location are encrypted

kafka-json-schema-console-consumer \
  --bootstrap-server localhost:6969 \
  --property schema.registry.url=http://localhost:8081 \
  --property print.key=true \
  --topic customers \
  --from-beginning \
  --max-messages 2 2>&1  /dev/null | grep '{' | jq

Adding interceptor decrypt​

Let's add the decrypt interceptor to decipher messages

step-10-decrypt-interceptor.json:

{
  "kind" : "Interceptor",
  "apiVersion" : "gateway/v2",
  "metadata" : {
    "name" : "decrypt"
  },
  "spec" : {
    "comment" : "Adding interceptor: decrypt",
    "pluginClass" : "io.conduktor.gateway.interceptor.DecryptPlugin",
    "priority" : 100,
    "config" : {
      "topic" : "customers",
      "schemaRegistryConfig" : {
        "host" : "http://schema-registry:8081"
      }
    }
  }
}
curl \
    --silent \
    --request PUT "http://localhost:8888/gateway/v2/interceptor" \
    --header "Content-Type: application/json" \
    --user "admin:conduktor" \
    --data @step-10-decrypt-interceptor.json | jq

Listing interceptors​

Listing interceptors on gateway1

curl \
    --silent \
    --request GET "http://localhost:8888/gateway/v2/interceptor" \
    --user "admin:conduktor" | jq

Let's make sure they are decrypted​

password and visa and the nested field address.location are decrypted

kafka-json-schema-console-consumer \
  --bootstrap-server localhost:6969 \
  --property schema.registry.url=http://localhost:8081 \
  --property print.key=true \
  --topic customers \
  --from-beginning \
  --max-messages 2 2>&1 | grep '{' | jq

Tearing down the docker environment​

Remove all your docker processes and associated volumes

  • --volumes: Remove named volumes declared in the "volumes" section of the Compose file and anonymous volumes attached to containers.
docker compose down --volumes

Conclusion

Yes, encryption in the Kafka world can be simple!