# Quick Start - Use CSFLE to Protect Sensitive Data This quick start shows you how to use CSFLE to protect sensitive data in Confluent Cloud on AWS, Azure, and Google Cloud. ## Requirements * Create a Confluent Cloud cluster with an example topic name of `test` or any topic name you choose. * [Stream Governance Advanced package](../../../stream-governance/packages.md#stream-gov-packages) must be enabled. * Create a Schema Registry cluster. For details, see [Quick Start for Schema Management on Confluent Cloud](../../../get-started/schema-registry.md#cloud-sr-config) and [Manage Schemas in Confluent Cloud](../../../sr/schemas-manage.md#sr-prv). ### Quick start using CSFLE Here is an example of CSFLE using the Avro data format for each of the KMS providers. ### Step 1 - Configure the KMS provider 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://docs.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. ```shell export LOCAL_SECRET=

``` ### Step 2 - Start the producer To start the producer, run the `kafka-avro-console-producer` command for the KMS provider that you want to use, where `` is the bootstrap URL for your Confluent Cloud cluster and `` is the URL for your Schema Registry instance. ### AWS ```shell ./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_API_KEY}:${SR_API_SECRET} \ --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 ```shell ./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_API_KEY}:${SR_API_SECRET} \ --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://myvault.vault.azure.net/keys/mykey/xxxx", "encrypt.kms.type": "azure-kms" }, "onFailure": "ERROR,NONE"}]}' ``` ### Google Cloud ```shell ./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_API_KEY}:${SR_API_SECRET} \ --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/myproject/locations/us-west1/keyRings/mykeyring/cryptoKeys/mykey", "encrypt.kms.type": "gcp-kms" }, "onFailure": "ERROR,NONE"}]}' ``` ### Hashicorp Vault ```shell ./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_API_KEY}:${SR_API_SECRET} \ --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": "hcvault-kek1", "encrypt.kms.key.id": "http://127.0.0.1:8200/transit/keys/my-key", "encrypt.kms.type": "hcvault" }, "onFailure": "ERROR,NONE"}]}' ``` ### Local key ```shell ./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_API_KEY}:${SR_API_SECRET} \ --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": "my-local-key", "encrypt.kms.type": "local-kms" }, "onFailure": "ERROR,NONE"}]}' ``` ### 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 Cloud cluster. ```shell ./bin/kafka-avro-console-consumer --bootstrap-server \ --topic test \ --property schema.registry.url= \ --consumer.config config.properties ``` After you run the producer and consumer, you can verify that the data is encrypted and decrypted by using the `kafka-configs --describe` command for the topic. ```shell kafka-configs --bootstrap-server \ --entity-type topics \ --entity-name test \ --describe ``` Example test record should look like this: ```text {"f1": "foo"} {"f1": "foo", "f2": {"string": "bar"}} ``` ## Related content * [Protect Sensitive Data Using Client-Side Field Level Encryption on Confluent Cloud](overview.md#csfle-overview) * [Configuration Parameters for Client-Side Field Level Encryption](config-parameters.md#config-parameters-csfle) * [Code Examples for Client-Side Field Level Encryption in Confluent Cloud](code-examples.md#code-examples-csfle)