# Manage Client-Side Encryption in Confluent Platform for Self-Managed Connectors [Client-Side Field Level Encryption (CSFLE)](../security/protect-data/csfle/overview.md#csfle-overview) and [client-side payload encryption (CSPE)](https://docs.confluent.io/platform/current/security/encrypt/cspe.html) are security features that allow you to safeguard sensitive data, such as personally identifiable information (PII), by enabling field-level or payload encryption both at the producer and consumer levels. By encrypting and decrypting individual fields or complete payload/message within your data, CSFLE or CSPE ensures that access to sensitive information is tightly controlled, granting only authorized stakeholders access to the data they are permitted to see. #### IMPORTANT For self-managed connectors on Confluent Platform, CSFLE or CSPE is supported only on Confluent Enterprise 8.0 and later. ## Limitations Note the following limitations: * The connector does not support automatic schema registration. You must manually register schemas before creating the connectors. * The connector only supports encryption for fields of type `string` or `bytes` for CSFLE. * CSFLE does not support `string` and `bytes` fields within nested JSON_SR formats when the `value.converter.decimal.format` is set to `BASE64`. To workaround this limitation, set `value.converter.decimal.format` to `NUMERIC`. * The [reporter topics](/platform/current/connect/userguide.html#kconnect-reporter) are not covered. Ensure that the error and success response do not contain any sensitive information while using reporter topics. ## Supported connectors The following table list the connector and its minimum version that support CSFLE. | Connector | Minimum supported version | |-----------------------------------------------------------------------------------------------------------------------------------------|-----------------------------| | [ActiveMQ Source](https://docs.confluent.io/kafka-connectors/activemq-source/current/index.html) | 12.2.9 | | [Amazon CloudWatch Metrics Sink](https://docs.confluent.io/kafka-connectors/aws-cloudwatch-metrics/current/index.html) | 2.0.1 | | [Amazon DynamoDB Sink](https://docs.confluent.io/kafka-connectors/aws-dynamodb/current/overview.html) | 1.4.1 | | [Amazon Kinesis Source](https://docs.confluent.io/kafka-connectors/kinesis/current/index.html) | 1.3.27 | | [Amazon Redshift Sink](https://docs.confluent.io/kafka-connectors/aws-redshift/current/overview.html) | 1.2.6 | | [Amazon S3 Sink](https://docs.confluent.io/kafka-connectors/s3-sink/current/index.html) | 10.6.0 | | [Amazon S3 Source](https://docs.confluent.io/kafka-connectors/s3-source/current/index.html) | 2.6.10 | | [Amazon SQS Source](https://docs.confluent.io/kafka-connectors/sqs/current/index.html) | 2.0.3 | | [Apache Kudu (Source and Sink)](https://docs.confluent.io/kafka-connect-kudu/current/index.html) | 1.0.5 | | [AWS Lambda Sink](https://docs.confluent.io/kafka-connectors/aws-lambda/current/index.html) | 2.0.10 | | [Azure Blob Storage Source](https://docs.confluent.io/kafka-connectors/azure-blob-storage-source/current/index.html) | 2.6.10 | | [Azure Blob Storage Sink](https://docs.confluent.io/kafka-connectors/azure-blob-storage-sink/current/overview.html) | 1.6.27 | | [Azure Data Lake Storage Gen2 Sink](https://docs.confluent.io/kafka-connectors/azure-data-lake-gen2-sink/current/overview.html) | 1.6.27 | | [Azure Cognitive Search Sink](https://docs.confluent.io/kafka-connectors/azure-search/current/index.html) | 1.1.7 | | Kafka Connect for Azure Cosmos DB (Source and Sink) | 1.17.0 | | [Azure Functions Sink](https://docs.confluent.io/kafka-connectors/azure-functions/current/index.html) | 2.0.4 | | [Azure Synapse Analytics Sink](https://docs.confluent.io/kafka-connectors/azure-sql-dw/current/index.html) | 1.0.9 | | [Google BigQuery Sink](https://docs.confluent.io/kafka-connectors/bigquery/current/index.html) | 2.5.7 | | [Cassandra Sink](https://docs.confluent.io/kafka-connectors/cassandra/current/index.html) | 2.0.10 | | [Data Diode (Source and Sink)](https://docs.confluent.io/kafka-connectors/data-diode/current/index.html) | 1.2.6 | | [Databricks Delta Lake Sink for AWS](https://docs.confluent.io/kafka-connectors/databricks-delta-lake-sink/current/) | 1.0.19 | | Datadog Logs Sink | 1.3.0 | | [Debezium connector for MySQL](https://docs.confluent.io/debezium-connect-mysql-source/current/index.html) | 2.4.2 | | [Debezium connector for PostgreSQL](https://docs.confluent.io/debezium-connect-postgres-source/current/index.html) | 2.5.4 | | [Debezium connector for SQL Server](https://docs.confluent.io/debezium-connect-sqlserver-source/current/index.html) | 2.5.4 | | [Elasticsearch Sink](https://docs.confluent.io/kafka-connectors/elasticsearch/current/index.html) | 14.1.2 | | [Google Cloud Functions Sink](https://docs.confluent.io/kafka-connectors/gcp-functions/current/index.html) | 1.2.4 | | [Google Cloud Storage Sink](https://docs.confluent.io/kafka-connectors/gcs-sink/current/index.html) | 10.2.1 | | [Google Cloud Storage Source](https://docs.confluent.io/kafka-connectors/gcs-source/current/index.html) | 10.2.1 | | [GitHub Source](https://docs.confluent.io/kafka-connectors/github/current/index.html) | 2.1.8 | | [Google Cloud Pub/Sub Source](https://docs.confluent.io/kafka-connectors/gcp-pubsub/current/index.html) | 1.2.9 | | [Google Cloud Spanner Sink](https://docs.confluent.io/kafka-connectors/gcp-spanner/current/index.html) | 1.0.16 | | [Google Firebase Realtime Database Connector (Source and Sink)](https://docs.confluent.io/kafka-connectors/firebase/current/index.html) | 1.2.6 | | [HDFS 3 Source](https://docs.confluent.io/kafka-connectors/hdfs3-source/current/index.html) | 2.6.10 | | [HDFS 3 Sink](https://docs.confluent.io/kafka-connectors/hdfs3-sink/current/index.html) | 1.2.4 | | [HEAVY-AI (formerly OmniSci) Sink](https://docs.confluent.io/kafka-connectors/heavy-ai/current/index.html) | 1.0.9 | | [HTTP Sink](https://docs.confluent.io/kafka-connectors/http/current/index.html) | 1.7.8 | | [HTTP Source](https://docs.confluent.io/kafka-connectors/http-source/current/index.html) | 0.2.5 | | [IBM MQ Sink](https://docs.confluent.io/kafka-connectors/ibmmq-sink/current/index.html) | 2.1.15 | | [IBM MQ Source](https://docs.confluent.io/kafka-connectors/ibmmq-source/current/index.html) | 12.2.9 | | [InfluxDB Source](https://docs.confluent.io/kafka-connectors/influxdb/current/influx-db-source-connector/overview.html) | 1.2.11 | | [JDBC (Source and Sink)](https://docs.confluent.io/kafka-connectors/jdbc/current/index.html) | 10.8.2 | | [Jira Source](https://docs.confluent.io/kafka-connectors/jira/current/overview.html) | 1.2.13 | | [JMS Sink Connector](https://docs.confluent.io/kafka-connectors/jms-sink/current/index.html) | 2.1.15 | | [JMS Source Connector](https://docs.confluent.io/kafka-connect-jms-source/current/index.html) | 12.2.9 | | MongoDB Atlas Sink | 1.15.0 | | [Netezza Sink Connector](https://docs.confluent.io/kafka-connect-netezza/current/index.html) | 1.0.7 | | [Oracle XStream CDC Source](https://docs.confluent.io/kafka-connectors/oracle-xstream-cdc-source/current/overview.html) | 1.1.0 | | [PagerDuty Sink [Deprecated]](https://docs.confluent.io/kafka-connect-pagerduty-sink/current/index.html) | 1.0.10 | | [Redis Sink](https://docs.confluent.io/kafka-connect-redis/current/index.html) | 0.0.8 | | [Salesforce Bulk API (Source and Sink)](https://docs.confluent.io/kafka-connect-salesforce-bulk-api/current/index.html) | 2.0.25 | | [Salesforce (Source and Sink)](https://docs.confluent.io/kafka-connect-salesforce/current/index.html) | 2.0.25 | | [ServiceNow (Source and Sink)](https://docs.confluent.io/kafka-connect-servicenow/current/index.html) | 2.5.4 | | [SFTP (Source and Sink)](https://docs.confluent.io/kafka-connect-sftp/current/index.html) | 3.2.11 | | [SNMP Trap Source](https://docs.confluent.io/kafka-connect-snmp/current/index.html) | 1.3.2 | | Snowflake Sink | 3.1.1 | | [Solace Source](https://docs.confluent.io/kafka-connectors/solace/source/current/overview.html) | 1.2.8 | | [Solace Sink](https://docs.confluent.io/kafka-connectors/solace/sink/current/overview.html) | 2.1.15 | | [Splunk S2S Source](https://docs.confluent.io/kafka-connectors/splunk-s2s/current/overview.html) | 2.2.1 | | [Splunk Source Connector](https://docs.confluent.io/kafka-connect-splunk-source/current/index.html) | 1.1.5 | | [Kafka Connect Spooldir](https://docs.confluent.io/kafka-connect-spooldir/current/index.html) | 2.0.67 | | [Syslog Source](https://docs.confluent.io/kafka-connect-syslog/current/index.html) | 1.5.10 | | [TIBCO EMS Sink](https://docs.confluent.io/kafka-connect-tibco/current/sink/index.html) | 2.1.15 | | [TIBCO EMS Source](https://docs.confluent.io/kafka-connect-tibco/current/source/index.html) | 1.2.9 | | [Vertica Sink](https://docs.confluent.io/kafka-connect-vertica/current/overview.html) | 1.3.2 | | [VMware Tanzu Gemfire Sink](https://docs.confluent.io/kafka-connectors/gemfire/current/index.html) | 1.0.18 | | [Zendesk Source Connector](https://docs.confluent.io/kafka-connect-zendesk/current/index.html) | 1.3.4 | ## Requirements To use CSFLE or CSPE in Confluent Platform with self-managed connectors, you must meet the following requirements: - An installation of Confluent Enterprise 8.0 and later with the CSFLE or CSPE Add-On enabled. - Ensure Schema Registry is configured with the following properties before it starts: ```shell resource.extension.class=io.confluent.kafka.schemaregistry.rulehandler.RuleSetResourceExtension,io.confluent.dekregistry.DekRegistryResourceExtension confluent.license= confluent.license.addon.csfle= ``` #### NOTE The value for `confluent.license.addon.csfle` is the same as your main `confluent.license` key. - An external KMS to manage your Key Encryption Keys (KEKs). For more information, see [Manage KEKs](../security/protect-data/csfle/manage-keys.md#manage-keks-csfle). - The [KMS provider](/platform/current/security/protect-data/csfle/quick-start.html#step-1-configure-the-kms-provider) must be configured for the connector. - A Kafka topic to use as a data source or destination. ## Manage client-side encryption At a high level, you can manage client-side encryption for self-managed connectors using the following 2-step process: 1. [Configure CSFLE in Confluent Platform](#configure-csfle-without-sharing-kek-cp) or [Configure CSPE in Confluent Platform](#configure-cspe-without-sharing-kek-cp) 2. [Enable CSFLE or CSPE in connectors](#enable-csfle-for-connectors-cp) ### Configure CSFLE in Confluent Platform CSFLE is available in Confluent Enterprise 8.0 and later to help you protect sensitive data in your Confluent Platform account and perform stream processing on encrypted data. You must configure CSFLE in Confluent Platform before you modify an existing self-managed connector or create a new one with CSFLE enabled. Complete the following configuration steps: - Define the [schema for the topic](../schema-registry/schema.md#c3-schema-create) and add tags to the fields in the schema that you want to encrypt. For more information, see [add tags to the schema fields](../security/protect-data/csfle/client-side.md#configure-csfle-schema-tags). - Define an encryption policy that specifies rules to use to encrypt the tags. For more information, see [add encryption rules](../security/protect-data/csfle/client-side.md#configure-csfle-encryption-rule) - Grant DeveloperWrite permission for encryption key. - Add the following parameters in the connector configuration: ### AWS CSFLE Rule Executor For AWS, pass the following configuration parameters: | Parameter | Description | |------------------------------------------------------|--------------------------------| | `rule.executors._default_.param.access.key.id=?` | The AWS access key identifier. | | `rule.executors._default_.param.secret.access.key=?` | The AWS secret access key. | ### Azure CSFLE Rule Executor For Azure, pass the following configuration parameters: | Parameter | Description | |------------------------------------------------|------------------------------| | `rule.executors._default_.param.tenant.id` | The Azure tenant identifier. | | `rule.executors._default_.param.client.id` | The Azure client identifier. | | `rule.executors._default_.param.client.secret` | The Azure client secret. | ### Google Cloud CSFLE Rule Executor For Google Cloud, pass the following configuration parameters: | Parameter | Description | |-------------------------------------------------|--------------------------------------------------------| | `rule.executors._default_.param.account.type` | This parameter contains the Google Cloud account type. | | `rule.executors._default_.param.client.id` | The Google Cloud client identifier. | | `rule.executors._default_.param.client.email` | The Google Cloud client email address. | | `rule.executors._default_.param.private.key.id` | The Google Cloud private key identifier. | | `rule.executors._default_.param.private.key` | The Google Cloud private key. | ### HashiCorp Vault CSFLE Rule Executor For HashiCorp Vault, pass the following configuration parameters: | Parameter | Description | |--------------------------------------------|----------------------------------------------------------| | `rule.executors._default_.param.token.id` | The token identifier for HashiCorp Vault. | | `rule.executors._default_.param.namespace` | The namespace for HashiCorp Vault Enterprise (optional). | For more information, see [Configure CSFLE for use with Confluent Enterprise](../security/protect-data/csfle/client-side.md#configure-csfle). ### Configure CSPE in Confluent Platform CSPE is available in Confluent Enterprise 8.0 and later to help you protect sensitive data in your Confluent Platform account and perform stream processing on encrypted data. You must configure CSPE in Confluent Platform before you modify an existing self-managed connector or create a new one with CSPE enabled. Complete the following configuration steps: - Define the [schema for the topic](../schema-registry/schema.md#c3-schema-create) that you want to encrypt. - Define an encryption policy that specifies rules for encoding. For more information, see [add encoding rules](https://docs.confluent.io/platform/current/security/encrypt/cspe.html). - Grant DeveloperWrite permission for encryption key. - Add the following parameters in the connector configuration: ### AWS CSPE Rule Executor For AWS, pass the following configuration parameters: | Parameter | Description | |------------------------------------------------------|--------------------------------| | `rule.executors._default_.param.access.key.id=?` | The AWS access key identifier. | | `rule.executors._default_.param.secret.access.key=?` | The AWS secret access key. | ### Azure CSPE Rule Executor For Azure, pass the following configuration parameters: | Parameter | Description | |------------------------------------------------|------------------------------| | `rule.executors._default_.param.tenant.id` | The Azure tenant identifier. | | `rule.executors._default_.param.client.id` | The Azure client identifier. | | `rule.executors._default_.param.client.secret` | The Azure client secret. | ### Google Cloud CSPE Rule Executor * **sync:** google-cloud-cspe-rule-executor For Google Cloud, pass the following configuration parameters: | Parameter | Description | |-------------------------------------------------|--------------------------------------------------------| | `rule.executors._default_.param.account.type` | This parameter contains the Google Cloud account type. | | `rule.executors._default_.param.client.id` | The Google Cloud client identifier. | | `rule.executors._default_.param.client.email` | The Google Cloud client email address. | | `rule.executors._default_.param.private.key.id` | The Google Cloud private key identifier. | | `rule.executors._default_.param.private.key` | The Google Cloud private key. | ### HashiCorp Vault CSPE Rule Executor For HashiCorp Vault, pass the following configuration parameters: | Parameter | Description | |--------------------------------------------|----------------------------------------------------------| | `rule.executors._default_.param.token.id` | The token identifier for HashiCorp Vault. | | `rule.executors._default_.param.namespace` | The namespace for HashiCorp Vault Enterprise (optional). | For more information, see [Configure CSPE for use with Confluent Enterprise](https://docs.confluent.io/platform/current/security/encrypt/cspe.html) ### Enable CSFLE or CSPE in connectors To enable CSFLE or CSPE for connectors, define the following parameters with the mentioned boolean values in the connector configuration: #### NOTE If you do not add these values in the connector configuration, CSFLE or CSPE might not work properly. * `csfle.enabled=true` * `value.converter.auto.register.schemas=false` * `value.converter.use.latest.version=true` * `key.converter.auto.register.schemas=false` * `key.converter.use.latest.version=true` #### NOTE - To fetch the latest value schema from Schema Registry, use `value.converter.latest.cache.ttl.sec`, that allows you to define the time interval, in seconds, after which the connector fetches the latest version of the value schema. By default, its value is set to `-1`. To enable it, enter the desired time interval in seconds for this parameter. - Similar to the value schema, use `key.converter.latest.cache.ttl.sec` to define the time interval, in seconds, after which the converter fetches the latest key schema from Schema Registry. The default value is `-1`. Change this value to the desired time interval in seconds.