# 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](overview.md#csfle-overview). ## 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](#csfle-quick-start-config-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. ```properties bootstrap.servers= security.protocol=SASL_SSL sasl.mechanism=PLAIN sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="" \ 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`: ```shell $CONFLUENT_HOME/bin/kafka-topics --bootstrap-server \ --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](manage-keys.md#create-kek-csfle) and [Register a KEK](manage-keys.md#register-kek-csfle). 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. ### AWS KMS 1. Configure AWS KMS. For more information, see [AWS KMS](https://docs.aws.amazon.com/kms/latest/developerguide/create-keys.html). 2. Add the environment variables for your AWS credentials to both the producer and consumer. ```shell export AWS_ACCESS_KEY_ID=XXXX export AWS_SECRET_ACCESS_KEY=XXXX ``` ### Azure Key Vault 1. Configure Azure Key Vault. For more information, see [Azure Key Vault](https://learn.microsoft.com/en-us/azure/key-vault/general/overview). 2. Add the environment variables for your Azure credentials to both the producer and consumer. ```shell export AZURE_TENANT_ID=XXXX export AZURE_CLIENT_ID=XXXX export AZURE_CLIENT_SECRET=XXXX ``` ### Google Cloud KMS 1. Configure Google Cloud KMS. For more information, see [Google Cloud KMS](https://cloud.google.com/kms/docs/quickstart). 2. Add the environment variables for your Google Cloud credentials to both the producer and consumer. ```shell export GOOGLE_APPLICATION_CREDENTIALS=PATH_TO_CREDS.json ``` ### HashiCorp Vault 1. Configure HashiCorp Vault. For more information, see [HashiCorp Vault - Getting Started](https://developer.hashicorp.com/vault#getting-started). 2. Add the environment variables for your HashiCorp Vault credentials to both the producer and consumer. ```shell export VAULT_TOKEN=dev-only-token ``` ### Local key 1. Create a local key. 2. Add the environment variables for your local key credentials to the producer and consumer. ```shell export LOCAL_SECRET=

``` ### 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. ```shell export SR_USERNAME= export SR_PASSWORD= ``` To start the producer, run the `kafka-avro-console-producer` command for the KMS provider that you configured in [Step 1](#csfle-quick-start-step1), where `` is the bootstrap URL for your Confluent Platform cluster and `` 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. ### AWS KMS ```shell $CONFLUENT_HOME/bin/kafka-avro-console-producer --bootstrap-server \ --property 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"}]}' ``` ### Azure Key Vault ```shell $CONFLUENT_HOME/bin/kafka-avro-console-producer --bootstrap-server \ --property 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.azure.net/keys//", "encrypt.kms.type": "azure-kms" }, "onFailure": "ERROR,NONE"}]}' ``` ### Google Cloud KMS ```shell $CONFLUENT_HOME/bin/kafka-avro-console-producer --bootstrap-server \ --property 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//locations//keyRings//cryptoKeys/", "encrypt.kms.type": "gcp-kms" }, "onFailure": "ERROR,NONE"}]}' ``` ### HashiCorp Vault ```shell $CONFLUENT_HOME/bin/kafka-avro-console-producer --bootstrap-server \ --property 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:///transit/keys/", "encrypt.kms.type": "hcvault" }, "onFailure": "ERROR,NONE"}]}' ``` ### Local key ```shell $CONFLUENT_HOME/bin/kafka-avro-console-producer --bootstrap-server \ --property 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": "", "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](client-side.md#configure-csfle-schema-tags). * `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](client-side.md#configure-csfle-encryption-rule) and [KEK parameters](manage-keys.md#kek-parameters-csfle). After the producer starts, enter a record in Avro JSON format at the prompt and press **Enter** to publish it: ```text {"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 `` 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](#csfle-quick-start-step2). ```shell $CONFLUENT_HOME/bin/kafka-avro-console-consumer --bootstrap-server \ --topic test \ --from-beginning \ --property 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: ```text {"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](#csfle-quick-start-step2). Do not export the KMS credentials from [Step 1](#csfle-quick-start-step1): ```shell export SR_USERNAME= export SR_PASSWORD= ``` Run the same consumer command: ```shell $CONFLUENT_HOME/bin/kafka-avro-console-consumer --bootstrap-server \ --topic test \ --from-beginning \ --property 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`: ```text {"f1": ""} ``` Comparing this output with [Step 3](#csfle-quick-start-step3) confirms that CSFLE encrypted the `f1` field. ## Related content * [Protect Sensitive Data Using Client-Side Field Level Encryption on Confluent Platform](overview.md#csfle-overview) * [Configure Client-Side Field Level Encryption on Confluent Platform](config-parameters.md#config-parameters-csfle) * [Client-Side Field Level Encryption Examples for Confluent Platform](code-examples.md#code-examples-csfle)