Quick Start - Protect Sensitive Data on Confluent Platform with CSFLE

This quick start shows you how to use client-side field level encryption (CSFLE) to protect sensitive data on Confluent Platform.

CSFLE encrypts individual fields in your data on the client before the data is sent to Kafka. It uses a key encryption key (KEK) stored in your key management service (KMS) to protect the data encryption keys (DEKs) that encrypt your field data. To learn about these concepts before you begin, see Protect Sensitive Data Using Client-Side Field Level Encryption on Confluent Platform.

Requirements

Before you begin, ensure you have:

  • A running Confluent Enterprise cluster with:

    • A topic named test (or your chosen topic name)

    • Schema Registry configured and running

    • The necessary credentials to access your cluster:

      • Bootstrap server URL

      • Schema Registry URL

      • Schema Registry username and password

  • Access to the Confluent CLI from a terminal where you’ll run the commands.

  • The required configuration file config.properties in your working directory. For the contents of this file, see Create the configuration file.

Create the configuration file

The producer and consumer commands in this quick start use a config.properties file to connect to your Kafka cluster. Create this file in your working directory with the following properties, replacing the placeholder values with the credentials for your cluster.

bootstrap.servers=<bootstrap-url>
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="<kafka-username>" \
  password="<kafka-password>";

Where:

  • bootstrap.servers is the bootstrap URL for your Confluent Platform cluster.

  • security.protocol is the protocol used to communicate with brokers. Use SASL_SSL when TLS/SSL encryption is enabled, or SASL_PLAINTEXT otherwise.

  • sasl.mechanism is the SASL mechanism used for client connections.

  • sasl.jaas.config provides the credentials that the producer and consumer use to authenticate to the Kafka brokers.

Note

The producer and consumer commands pass the Schema Registry URL and credentials directly on the command line, so you don’t need to include them in config.properties. Adjust the broker connection properties to match the authentication method configured for your cluster.

Before you run the producer and consumer commands, make sure the topic you plan to use already exists. If the test topic doesn’t exist, create it with kafka-topics:

$CONFLUENT_HOME/bin/kafka-topics --bootstrap-server <bootstrap-url> \
  --command-config config.properties \
  --create --topic test

Quick start using CSFLE

In this example you use CSFLE with the Avro data format for each of the key management service (KMS) providers. A KMS is a service that securely creates, stores, and manages the encryption keys used to protect sensitive data. It acts as a trusted source for encryption/decryption keys, allowing applications to encrypt data before sending it to Confluent Enterprise and decrypt it when reading.

Step 1 - Configure the KMS provider

Before you can encrypt data, you need a KEK in your KMS. The following steps configure access to the KMS provider that you want to use. For the steps to create and register the KEK itself, see Create a KEK and Register a KEK. If you don’t register the KEK beforehand, the client registers it on demand when you run the producer.

Complete the steps for the KMS provider that you want to use.

  1. Configure AWS KMS. For more information, see AWS KMS.

  2. Add the environment variables for your AWS credentials to both the producer and consumer.

export AWS_ACCESS_KEY_ID=XXXX
export AWS_SECRET_ACCESS_KEY=XXXX

  1. Configure Azure Key Vault. For more information, see Azure Key Vault.

  2. Add the environment variables for your Azure credentials to both the producer and consumer.

export AZURE_TENANT_ID=XXXX
export AZURE_CLIENT_ID=XXXX
export AZURE_CLIENT_SECRET=XXXX

  1. Configure Google Cloud KMS. For more information, see Google Cloud KMS.

  2. Add the environment variables for your Google Cloud credentials to both the producer and consumer.

export GOOGLE_APPLICATION_CREDENTIALS=PATH_TO_CREDS.json

  1. Configure HashiCorp Vault. For more information, see HashiCorp Vault - Getting Started.

  2. Add the environment variables for your HashiCorp Vault credentials to both the producer and consumer.

    export VAULT_TOKEN=dev-only-token

  1. Create a local key.

  2. Add the environment variables for your local key credentials to the producer and consumer.

    export LOCAL_SECRET=<output of "openssl rand -base64 16">

Step 2 - Start the producer

Before you start the producer, export your Schema Registry username and password as environment variables. The producer command references these variables to authenticate to Schema Registry.

export SR_USERNAME=<schema-registry-username>
export SR_PASSWORD=<schema-registry-password>

To start the producer, run the kafka-avro-console-producer command for the KMS provider that you configured in Step 1, where <bootstrap-url> is the bootstrap URL for your Confluent Platform cluster and <schema-registry-url> is the URL for your Schema Registry instance. Follow the instructions for your KMS provider because the value.rule.set parameters differ by KMS provider.

$CONFLUENT_HOME/bin/kafka-avro-console-producer --bootstrap-server <bootstrap-url> \
  --property schema.registry.url=<schema-registry-url> \
  --topic test \
  --producer.config config.properties \
  --property basic.auth.credentials.source=USER_INFO \
  --property basic.auth.user.info=${SR_USERNAME}:${SR_PASSWORD} \
  --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"f1","type":"string","confluent:tags":["PII"]}]}' \
  --property value.rule.set='{ "domainRules": [ { "name": "encryptPII", "type": "ENCRYPT", "tags":["PII"], "params": { "encrypt.kek.name": "aws-kek1", "encrypt.kms.key.id": "arn:aws:kms:us-east-1:xxxx:key/xxxx", "encrypt.kms.type": "aws-kms" }, "onFailure": "ERROR,NONE"}]}'

$CONFLUENT_HOME/bin/kafka-avro-console-producer --bootstrap-server <bootstrap-url> \
  --property schema.registry.url=<schema-registry-url> \
  --topic test \
  --producer.config config.properties \
  --property basic.auth.credentials.source=USER_INFO \
  --property basic.auth.user.info=${SR_USERNAME}:${SR_PASSWORD} \
  --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"f1","type":"string","confluent:tags":["PII"]}]}' \
  --property value.rule.set='{ "domainRules": [ { "name": "encryptPII", "type": "ENCRYPT", "tags":["PII"], "params": { "encrypt.kek.name": "azure-kek1", "encrypt.kms.key.id": "https://<vault-name>.vault.azure.net/keys/<key-name>/<key-version>", "encrypt.kms.type": "azure-kms" }, "onFailure": "ERROR,NONE"}]}'

$CONFLUENT_HOME/bin/kafka-avro-console-producer --bootstrap-server <bootstrap-url> \
  --property schema.registry.url=<schema-registry-url> \
  --topic test \
  --producer.config config.properties \
  --property basic.auth.credentials.source=USER_INFO \
  --property basic.auth.user.info=${SR_USERNAME}:${SR_PASSWORD} \
  --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"f1","type":"string","confluent:tags":["PII"]}]}' \
  --property value.rule.set='{ "domainRules": [ { "name": "encryptPII", "type": "ENCRYPT", "tags":["PII"], "params": { "encrypt.kek.name": "gcp-kek1", "encrypt.kms.key.id": "projects/<project-id>/locations/<location>/keyRings/<key-ring>/cryptoKeys/<key-name>", "encrypt.kms.type": "gcp-kms" }, "onFailure": "ERROR,NONE"}]}'

$CONFLUENT_HOME/bin/kafka-avro-console-producer --bootstrap-server <bootstrap-url> \
  --property schema.registry.url=<schema-registry-url> \
  --topic test \
  --producer.config config.properties \
  --property basic.auth.credentials.source=USER_INFO \
  --property basic.auth.user.info=${SR_USERNAME}:${SR_PASSWORD} \
  --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"f1","type":"string","confluent:tags":["PII"]}]}' \
  --property value.rule.set='{ "domainRules": [ { "name": "encryptPII", "type": "ENCRYPT", "tags":["PII"], "params": { "encrypt.kek.name": "vault-kek1", "encrypt.kms.key.id": "https://<vault-address>/transit/keys/<key-name>", "encrypt.kms.type": "hcvault" }, "onFailure": "ERROR,NONE"}]}'

$CONFLUENT_HOME/bin/kafka-avro-console-producer --bootstrap-server <bootstrap-url> \
  --property schema.registry.url=<schema-registry-url> \
  --topic test \
  --producer.config config.properties \
  --property basic.auth.credentials.source=USER_INFO \
  --property basic.auth.user.info=${SR_USERNAME}:${SR_PASSWORD} \
  --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"f1","type":"string","confluent:tags":["PII"]}]}' \
  --property value.rule.set='{ "domainRules": [ { "name": "encryptPII", "type": "ENCRYPT", "tags":["PII"], "params": { "encrypt.kek.name": "local-kek1", "encrypt.kms.key.id": "<local-key-id>", "encrypt.kms.type": "local-kms" }, "onFailure": "ERROR,NONE"}]}'

This command uses the following CSFLE properties:

  • value.schema defines the Avro schema for the data. The confluent:tags attribute marks the f1 field with the PII tag so that the encryption rule can target it. For more information, see Step 3: Add tags to the schema fields.

  • value.rule.set defines the encryption policy that encrypts the tagged fields:

    • encrypt.kek.name is the name of the KEK to use.

    • encrypt.kms.key.id is the key identifier in your KMS. Its format depends on your KMS provider.

    • encrypt.kms.type is the KMS type: aws-kms, azure-kms, gcp-kms, hcvault, or local-kms.

    For more information, see Step 4: Define an encryption policy and KEK parameters.

After the producer starts, enter a record in Avro JSON format at the prompt and press Enter to publish it:

{"f1": "foo"}

Step 3 - Start the consumer with decryption

To start the consumer with decryption, run the kafka-avro-console-consumer command for the KMS provider that you want to use, where <bootstrap-url> is the bootstrap URL for your Confluent Platform cluster. Like the producer, the consumer passes the Schema Registry URL and credentials on the command line so it can fetch the schema and the DEK that it needs to decrypt the data. The --from-beginning option ensures the consumer reads the record that you published in Step 2.

$CONFLUENT_HOME/bin/kafka-avro-console-consumer --bootstrap-server <bootstrap-url> \
  --topic test \
  --from-beginning \
  --property schema.registry.url=<schema-registry-url> \
  --consumer.config config.properties \
  --property basic.auth.credentials.source=USER_INFO \
  --property basic.auth.user.info=${SR_USERNAME}:${SR_PASSWORD}

Because this consumer has access to your KMS, it decrypts the f1 field and prints the original record:

{"f1": "foo"}

Step 4 - Verify that the data is encrypted

To confirm that the data is stored encrypted, consume the topic again from a client without KMS access.

In a separate terminal, export only the Schema Registry credentials from Step 2. Do not export the KMS credentials from Step 1:

export SR_USERNAME=<schema-registry-username>
export SR_PASSWORD=<schema-registry-password>

Run the same consumer command:

$CONFLUENT_HOME/bin/kafka-avro-console-consumer --bootstrap-server <bootstrap-url> \
  --topic test \
  --from-beginning \
  --property schema.registry.url=<schema-registry-url> \
  --consumer.config config.properties \
  --property basic.auth.credentials.source=USER_INFO \
  --property basic.auth.user.info=${SR_USERNAME}:${SR_PASSWORD}

Without KMS access, this consumer can’t decrypt the f1 field. The output shows the encrypted value instead of foo:

{"f1": "<encrypted-value>"}

Comparing this output with Step 3 confirms that CSFLE encrypted the f1 field.

Related content