<a id="use-client-side-payload-encryption"></a>

# Protect Sensitive Data Using Client-Side Payload Encryption on Confluent Platform

Use Client-Side Payload Encryption (CSPE) to easily encrypt entire messages at the payload or message level.
This feature provides comprehensive security assurance without the operational burden of
field-by-field encryption.

Features
: - Encrypt the entire message payload with a single rule, reducing the operational
    complexity and risk of exposing sensitive data.
  - Meet stringent compliance requirements by ensuring all data is encrypted by default,
    minimizing data classification burdens.
  - Maintain uniform encryption practices across your platforms.
  - Bring all of your data, even the “all sensitive” payloads without clear schemas,
    into Confluent Platform.
  <br/>
  With CSPE you can define an encryption rule at the schema level to encrypt the entire message payload.
  <br/>
  CSPE is enabled by specifying an encoding rule. Domain rules are run before serialization and after deserialization.
  Encoding rules are run after serialization and before deserialization.

Envelope encryption
: CSPE and Client-Side Field Level Encryption use envelope encryption.
  <br/>
  - A Data Encryption Key (DEK) is used to encrypt the plaintext data.
  - A Key Encryption Key (KEK) is used to encrypt the DEK. The KEK is required to decrypt the data.

Key ownership and management
: - Key Encryption Keys (KEKs): Users retain ownership and management of their KEKs (master keys) in their external
    Key Management System (KMS). Confluent interacts with the KEK via API calls for encryption/decryption of the DEK but
    never directly accesses or persists the KEKs. Customers can rotate KEKs through their KMS.
  - Data Encryption Keys (DEKs): DEKs are generated and managed by Confluent, scoped by subject, and stored as encrypted DEKs
    within the Schema Registry instance.

KEK access options and impact on managed services
: You must decide whether to grant Confluent access to your KEK. This decision determines the functionality available from fully managed services.
  <br/>
  - No Confluent access to KEK
    - Access level: No user or application in Confluent Cloud can access the encrypted data in plaintext.
    - Service limitation: Managed services like Flink or Connect cannot access the unencrypted data. Stream processing on encrypted fields is not possible.
    - Client complexity: Users must configure producers and consumers to connect directly to the KMS to access KEKs.
  - Confluent granted access to KEK
    - Managed services enabled: Sharing the KEK means Confluent can facilitate encrypt and decrypt operations.
    - Future functionality: New features automatically enabled when the KEK is shared with Confluent.
    - Simpler client setup: Less configuration is required for producers and consumers.

Availability and packaging
: - CSPE is an entitlement available only as part of the package.
  - Client support: CSPE supports Avro, Protobuf, and JSON Schema formats. Supported client languages include Java, C#/.NET, Go, Python, JavaScript, and C++.
  - Configuration interface: Deployment is available through API with CLI and Terraform.

Client versions
: CSPE requires the following minimum client versions:
  <br/>
  - Kafka Java Client: Confluent Platform 8.0.1 or later, Confluent Platform 8.1.0 or later, or Confluent Platform 8.2.0 or
    later
  - Go Client: version 2.11.1 or later
  - .NET Client: version 2.11.1 or later
  - JavaScript Client: version 1.4.1 or later
  - Python Client: version 2.11.1 or later
  <br/>
  CSPE is not supported in Confluent Platform 7.x releases.

## CSPE Quick start

1. Create a KEK using your key management service (KMS).

   For the steps to create a KEK, see [Create a KEK](https://docs.confluent.io/cloud/current/security/encrypt/csfle/manage-keys.html#create-kek-csflecreate-kek-csfle).
2. Register the KEK with the DEK Registry using Confluent Cloud Console, Confluent CLI, or REST APIs.

   Note that if the KEK is not registered beforehand, you can register it on demand
   by the client, assuming that the client has the appropriate permissions with the
   DEK Registry.

   For the steps to register a KEK with the DEK Registry, see
   [Register a KEK with the DEK Registry](https://docs.confluent.io/cloud/current/security/encrypt/csfle/manage-keys.html#register-kek-csfle).
3. Define an encryption policy that specifies
   rules for which tags use for encryption. The encryption policy is defined in
   a JSON file that is then uploaded to Confluent Platform. Here is an example of an
   encryption policy:
   ```json
   {
     "ruleSet": {
       "encodingRules": [
         {
           "name": "encryptPayload",
           "kind": "TRANSFORM",
           "type": "ENCRYPT_PAYLOAD",
           "mode": "WRITEREAD",
           "params": {
             "encrypt.kek.name": "<kek-name>"
           },
           "onFailure": "ERROR,NONE"
         }
       ]
     }
   }
   ```

   Note that you specified the name of the KEK in step 1. If the KEK has not yet
   been registered, you can optionally specify the KMS key ID and KMS type in the
   rule. The client automatically registers the KEK before registering any DEKs.
   ```json
   {
   "schema": "...",
   "metadata": {...},
   "ruleSet": {
       "encodingRules": [
         {
         "name": "encryptPayload",
         "kind": "TRANSFORM",
         "type": "ENCRYPT_PAYLOAD",
         "mode": "WRITEREAD",
         "tags": ["PII"],
         "params": {
             "encrypt.kek.name": "<kekName>",
             "encrypt.kms.key.id": "<kmsKeyId>",
             "encrypt.kms.type": "<kmsType>"
         }
         }
       ]
   }
   }
   ```

   During registration, if the schema is omitted, then the ruleset attaches
   to the latest schema in the subject.

   After registration, you need to include the following in the client:
   * `auto.register.schemas=false`
   * `use.latest.version=true`

   If you do not include these properties, the client attempts to register, or look up,
   a schema without any existing rules.

   You can specify these properties for DEKs in the rule parameters:

   `encrypt.dek.algorithm`
   : The encryption algorithm being used. Valid values include `AES128_GCM`,
     `AES256_GCM` (default), or `AES256_SIV`.

   `encrypt.dek.expiry.days`
   : If specified, automatic DEK rotation occurs. When a DEK is older than the
     expiration period, a new DEK is generated and used for new messages, while
     previous DEKs are available to decrypt older messages. There is a limit to
     the number of existing DEKs (10,000) that can be retained in the DEK Registry,
     so use this property cautiously.
4. Configure the KMS key encryption key. For Confluent Platform, the DEK Registry can be
   granted permission to access your KMS.

   To provide proper access for the DEK Registry, obtain the KMS key policy that
   needs to be added to the key, either through the Confluent Cloud Console or by using
   the following REST endpoint:
   ```text
   https://psrc-xxxxx.<region>.<provider>.confluent.cloud/dek-registry/v1/policy
   ```

   If the DEK Registry has not been granted permission to access the KMS, then the
   credentials to access the KMS must be specified on the client.

   For each of the supported KMS providers, the following dependencies are required,
   where `<cp-version>` is your Confluent Platform version:

   ### AWS

   ```xml
   <dependency>
      <groupId>io.confluent</groupId>
      <artifactId>kafka-schema-registry-client-encryption-aws</artifactId>
      <version><cp-version></version>
   </dependency>
   ```

   ### Azure

   ```xml
   <dependency>
      <groupId>io.confluent</groupId>
      <artifactId>kafka-schema-registry-client-encryption-azure</artifactId>
      <version><cp-version></version>
   </dependency>
   ```

   ### Google Cloud

   ```xml
   <dependency>
      <groupId>io.confluent</groupId>
      <artifactId>kafka-schema-registry-client-encryption-gcp</artifactId>
      <version><cp-version></version>
   </dependency>
   ```

   ### Hashicorp Vault

   ```xml
   <dependency>
      <groupId>io.confluent</groupId>
      <artifactId>kafka-schema-registry-client-encryption-hcvault</artifactId>
      <version><cp-version></version>
   </dependency>
   ```

   ### Local

   ```xml
   <dependency>
      <groupId>io.confluent</groupId>
      <artifactId>kafka-schema-registry-client-encryption</artifactId>
      <version><cp-version></version>
   </dependency>
   ```

   Next, you need to configure the following parameters on the clients.

   ### AWS

   ```text
   rule.executors._default_.param.access.key.id=<AWS access key>
   rule.executors._default_.param.secret.access.key=<AWS secret key>
   ```

   Alternatively, the AWS access key and AWS secret key can be passed
   using environment variables, named `AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY`.

   ### Azure

   ```text
   rule.executors._default_.param.tenant.id=<tenant ID>
   rule.executors._default_.param.client.id=<client ID>
   rule.executors._default_.param.client.secret=<secret value>
   ```

   ### Google Cloud

   ```text
   rule.executors._default_.param.client.id=<GCP Client ID>
   rule.executors._default_.param.client.email=<GCP Client Email>
   rule.executors._default_.param.private.key.id=<GCP Private Key ID>
   rule.executors._default_.param.private.key=<GCP Private Key Material>
   ```

   ### Hashicorp Vault

   ```text
   rule.executors._default_.param.token.id=root-token
   ```

   ### Local

   ```text
   rule.executors._default_.param.secret=<output of "openssl rand -base64 16">
   ```

   Now, whenever a message is sent, the `ssn` field is automatically encrypted
   before serialization and decrypted after deserialization.
5. Produce with Kafka serializers. Add the appropriate serializer to the producer
   properties:
   * KafkaAvroSerializer
   * KafkaProtobufSerializer
   * KafkaJsonSchemaSerializer

## Example

Here is an example of a rule set that contains a CSPE rule. Note that CSPE uses the `ENCRYPT_PAYLOAD` rule type,
rather than `ENCRYPT` rule type which CSFLE uses.

```none
{
"ruleSet": {
    "encodingRules": [
    {
        "name": "encryptPayload",
        "kind": "TRANSFORM",
        "type": "ENCRYPT_PAYLOAD",
        "mode": "WRITEREAD",
        "params": {
        "encrypt.kek.name": "local-kek1",
        "encrypt.kms.key.id": "mykey",
        "encrypt.kms.type": "local-kms"
        },
        "onFailure": "ERROR,NONE"
    }
    ]
}
}
```