<a id="co-gateway-governance"></a>

# Enforce Centralized Governance for Confluent Gateway using Confluent for Kubernetes

Configure Confluent Gateway deployed with Confluent for Kubernetes (CFK) to enforce Confluent Schema Registry data
contracts on the traffic between your Apache Kafka® clients and clusters. You
define policies once in Schema Registry, and Confluent Gateway enforces them on every producer and
consumer request without changes to your client applications.

For an overview of how centralized governance enforcement works, including the
supported checks and limitations, see the following topics:

* For Confluent Private Cloud Gateway, see [Enforce Data Governance Centrally
  with Confluent Gateway](https://docs.confluent.io/private-cloud-gateway/current/governance/gateway-governance.html).
* For the Confluent Gateway add-on for Confluent Platform, see [Enforce Data
  Governance Centrally with Confluent Gateway](https://docs.confluent.io/platform/current/gateway/governance/gateway-governance.html).

## Before you begin

Ensure your environment meets the following requirements:

* Confluent Gateway version 1.4 or later, deployed as described in
  [Configure and Deploy Confluent Gateway using Confluent for Kubernetes](co-gateway-deploy.md#co-gateway-deploy).
* CFK version 3.4 or later.
* For the Confluent Gateway add-on for Confluent Platform, a Confluent Enterprise license for Confluent Platform that
  includes the Confluent Gateway add-on. To use encryption rules, you also need the
  Client-Side Field Level Encryption (CSFLE) add-on license, which is a separate purchase on Confluent Platform.
  For more information, see
  [Confluent Platform licenses](https://docs.confluent.io/platform/current/installation/license.html).
* Confluent Schema Registry on Confluent Cloud, or Confluent Platform Schema Registry 8.4 or later.
* Client applications that produce and consume records through Confluent Gateway using
  Schema Registry serializers. Records must carry a schema ID, either in the Schema Registry wire
  format or as a schema globally unique identifier (GUID) in a record header.
* Network access from Confluent Gateway to Schema Registry and credentials for one of the
  supported authentication methods. For more information, see
  [Connect to Schema Registry](#co-gateway-governance-sr).
* For encryption, a key encryption key (KEK) registered in Schema Registry, Schema Registry
  configured to store encryption rules (the `RuleSetResourceExtension` and
  `DekRegistryResourceExtension` extensions and the client-side field level
  encryption license add-on), and permission for Confluent Gateway to use the key in
  your key management service (KMS). For more information, see
  [Configure encryption](#co-gateway-governance-encryption).

## Configure governance

Configure governance in the `Gateway` custom resource (CR) under
`spec.policies.dataGovernance`. The policy applies to traffic on all
Confluent Gateway routes.

```yaml
apiVersion: platform.confluent.io/v1beta1
kind: Gateway
metadata:
  name: confluent-gateway
  namespace: confluent
spec:
  policies:
    dataGovernance:
      schemaRegistry:
        urls:                      --- [1]
          - "https://<schema-registry-host>:8081"
        authentication: { ... }    --- [2]
        tls: { ... }               --- [3]
        ruleExecutors: [ ... ]     --- [4]
      key:
        enabled: false             --- [5]
      value:
        enabled: true              --- [5]
        subjectNameStrategy: TOPIC --- [6]
```

* [1] `schemaRegistry.urls`: The Schema Registry endpoints that Confluent Gateway uses to look
  up schemas and rules. Specify 1 to 16 URLs.
* [2] `schemaRegistry.authentication.type`: How Confluent Gateway authenticates to
  Schema Registry. Valid values are `basic` and `mtls`. For more information, see
  [Connect to Schema Registry](#co-gateway-governance-sr).
* [3] `schemaRegistry.tls`: The TLS settings for the Schema Registry connection. For
  more information, see [Connect to Schema Registry](#co-gateway-governance-sr).
* [4] `schemaRegistry.ruleExecutors`: The rule executors that run encryption
  rules. Required only for encryption. For more information, see
  [Configure encryption](#co-gateway-governance-encryption).
* [5] `key.enabled` and `value.enabled`: Enable or disable governance for
  record keys and record values. Both settings are required. When set to
  `false`, Confluent Gateway forwards that part of the record unchanged. Confluent Gateway
  evaluates keys and values independently, so you can govern only the part that
  has a schema.
* [6] `key.subjectNameStrategy` and `value.subjectNameStrategy`: How
  Confluent Gateway maps a record to a Schema Registry subject. Valid values are `TOPIC`
  (default), `RECORD`, and `TOPIC_RECORD`. Keys and values
  can use different strategies.

To pass additional Schema Registry client properties, such as timeouts, use
`schemaRegistry.additionalConfigs`.

#### NOTE
For JSON Schema, to have Confluent Gateway validate the payload against schema
constraints, such as `minimum` and `enum`, set
`json.fail.invalid.schema` to `"true"` in
`schemaRegistry.additionalConfigs`.

Confluent Gateway reads schemas and rules from Schema Registry. You define validation constraints
and encryption rules in the schemas, not in the Confluent Gateway CR.

<a id="co-gateway-governance-sr"></a>

## Connect to Schema Registry

Store Schema Registry credentials and certificates in Kubernetes secrets in the Confluent Gateway
namespace, and reference them from `schemaRegistry`.

Confluent Gateway supports basic and mutual TLS (mTLS) authentication to Schema Registry. Use mTLS
with Confluent Platform Schema Registry. Schema Registry on Confluent Cloud authenticates clients with API keys, so use
basic authentication.

### Basic authentication

Create a secret with a `basic.txt` key:

```yaml
apiVersion: v1
kind: Secret
metadata:
  name: sr-credentials
  namespace: confluent
type: Opaque
stringData:
  basic.txt: |
    username=<sr-username>
    password=<sr-password>
```

Reference the secret from the Confluent Gateway CR:

```yaml
spec:
  policies:
    dataGovernance:
      schemaRegistry:
        urls:
          - "http://schemaregistry.confluent.svc.cluster.local:8081"
        authentication:
          type: basic
          basic:
            secretRef: sr-credentials
```

If Schema Registry uses a certificate signed by a private certificate authority (CA), also
create a secret with a truststore that contains the CA certificate:

```yaml
apiVersion: v1
kind: Secret
metadata:
  name: sr-truststore
  namespace: confluent
type: Opaque
stringData:
  jksPassword.txt: jksPassword=<truststore-password>
data:
  truststore.jks: <base64-encoded-truststore>
```

Enable TLS, reference the truststore, and use an `https://` URL. If Schema Registry uses
a publicly trusted certificate, omit `secretRef`:

```yaml
spec:
  policies:
    dataGovernance:
      schemaRegistry:
        urls:
          - "https://schemaregistry.confluent.svc.cluster.local:8081"
        authentication:
          type: basic
          basic:
            secretRef: sr-credentials
        tls:
          enabled: true
          secretRef: sr-truststore
```

### Mutual TLS (mTLS)

Create a single secret with the Confluent Gateway client keystore and the truststore:

```yaml
apiVersion: v1
kind: Secret
metadata:
  name: sr-tls
  namespace: confluent
type: Opaque
stringData:
  jksPassword.txt: jksPassword=<store-password>
data:
  keystore.jks: <base64-encoded-keystore>
  truststore.jks: <base64-encoded-truststore>
```

You don’t need a credentials secret. Set the authentication type to `mtls`
and enable TLS:

```yaml
spec:
  policies:
    dataGovernance:
      schemaRegistry:
        urls:
          - "https://schemaregistry.confluent.svc.cluster.local:8081"
        authentication:
          type: mtls
        tls:
          enabled: true
          secretRef: sr-tls
```

## Configure governance per topic

Use `topics` to override the governance settings for specific topics. Each
entry must set at least one setting under `key` or `value`, and Confluent Gateway
applies the global value to any setting you omit. You can add up to 200 topics.

In the following example, governance is enabled globally for values, but
Confluent Gateway bypasses governance for the `public-events` topic and uses the
`RECORD` subject name strategy for the `inventory` topic:

```yaml
spec:
  policies:
    dataGovernance:
      key:
        enabled: false
      value:
        enabled: true
      topics:
        - name: public-events
          value:
            enabled: false
        - name: inventory
          value:
            subjectNameStrategy: RECORD
```

<a id="co-gateway-governance-encryption"></a>

## Configure encryption

To encrypt and decrypt records on Confluent Gateway, complete the following steps:

1. Configure Schema Registry to store encryption rules and keys. For more information,
   see [Configure data contract rules and encryption](https://docs.confluent.io/private-cloud-gateway/current/governance/gateway-governance-config.html#gateway-governance-data-contracts)
   for Confluent Private Cloud Gateway or [Configure data contract rules and
   encryption](https://docs.confluent.io/platform/current/gateway/governance/gateway-governance-config.html#gateway-governance-data-contracts)
   for the Confluent Gateway add-on for Confluent Platform.
2. Add rule executors and give Confluent Gateway access to the KMS, as described in
   [Configure KMS access](#co-gateway-governance-kms).
3. Add an encryption rule with `enableAt: GATEWAY` to the schema, as
   described in [Define encryption rules in the schema](#co-gateway-governance-rules).
4. Enable governance for the key or value that contains the data to encrypt.

<a id="co-gateway-governance-kms"></a>

### Configure KMS access

To run encryption rules, Confluent Gateway needs a rule executor and permission to use
the KMS key. Register `FieldEncryptionExecutor` for field-level encryption
rules, and `EncryptionExecutor` for full-payload encryption rules.

The following example uses AWS KMS on Amazon EKS with IAM Roles for Service
Accounts (IRSA):

1. Annotate the Confluent Gateway service account with an IAM role:
   ```yaml
   apiVersion: v1
   kind: ServiceAccount
   metadata:
     name: default
     namespace: confluent
     annotations:
       eks.amazonaws.com/role-arn: arn:aws:iam::<aws-account-id>:role/<irsa-role-name>
   ```
2. Confluent Gateway assumes the role that has access to the KMS key when it calls
   AWS KMS. Create a secret with the Amazon Resource Name (ARN) of that role:
   ```yaml
   apiVersion: v1
   kind: Secret
   metadata:
     name: aws-kms-role-params
     namespace: confluent
   type: Opaque
   stringData:
     role.arn: arn:aws:iam::<aws-account-id>:role/<kms-access-role-name>
   ```
3. Add the rule executors to the Schema Registry configuration:
   ```yaml
   spec:
     policies:
       dataGovernance:
         schemaRegistry:
           urls:
             - "http://schemaregistry.confluent.svc.cluster.local:8081"
           authentication:
             type: basic
             basic:
               secretRef: sr-credentials
           ruleExecutors:
             - name: encryptField
               class: io.confluent.kafka.schemaregistry.encryption.FieldEncryptionExecutor
               secretRef: aws-kms-role-params
             - name: encryptPayload
               class: io.confluent.kafka.schemaregistry.encryption.EncryptionExecutor
               secretRef: aws-kms-role-params
   ```

The preceding example annotates the `default` service account. If you set a
different service account in `spec.podTemplate.serviceAccountName`, annotate
that one instead.

Grant the IRSA role permission to assume the KMS access role, and grant the KMS
access role permission to encrypt and decrypt with the KMS key that your KEK
references.

Confluent Gateway also supports Azure Key Vault, Google Cloud KMS, and HashiCorp Vault.
To set up a KEK and KMS access for those providers, see [Configure the KMS
provider](https://docs.confluent.io/platform/current/security/protect-data/csfle/quick-start.html#step-1-configure-the-kms-provider).

<a id="co-gateway-governance-rules"></a>

### Define encryption rules in the schema

A schema’s rule set includes an `enableAt` setting that determines which
component runs its encryption rules. Set `enableAt` to `GATEWAY` so that
only Confluent Gateway runs the rules. For all valid values, see [Choose where
encryption runs](https://docs.confluent.io/private-cloud-gateway/current/governance/gateway-governance-config.html#choose-where-encryption-runs)
for Confluent Private Cloud Gateway or [Choose where encryption runs](https://docs.confluent.io/platform/current/gateway/governance/gateway-governance-config.html#choose-where-encryption-runs)
for the Confluent Gateway add-on for Confluent Platform.

To encrypt tagged fields, add a field-level rule. In the following example,
Confluent Gateway encrypts the fields tagged `PII`:

```json
{
  "schema": "...",
  "ruleSet": {
    "domainRules": [ {
      "name": "encryptPII",
      "kind": "TRANSFORM",
      "type": "ENCRYPT",
      "mode": "WRITEREAD",
      "tags": ["PII"],
      "params": {
        "encrypt.kek.name": "my-kek",
        "encrypt.kms.type": "aws-kms",
        "encrypt.kms.key.id": "<kms-key-arn>"
      },
      "onFailure": "ERROR,NONE"
    } ],
    "enableAt": "GATEWAY"
  }
}
```

With `onFailure` set to `ERROR,NONE`, Confluent Gateway raises an error if
encryption fails on the produce path. If decryption fails on the fetch path,
Confluent Gateway returns the record with the fields still encrypted and doesn’t raise
an error.

To encrypt the entire record value, add a payload rule:

```json
{
  "schema": "...",
  "ruleSet": {
    "encodingRules": [ {
      "name": "encryptPayload",
      "kind": "TRANSFORM",
      "mode": "WRITEREAD",
      "type": "ENCRYPT_PAYLOAD",
      "params": { "encrypt.kek.name": "my-kek" },
      "onSuccess": "NONE",
      "onFailure": "ERROR"
    } ],
    "enableAt": "GATEWAY"
  }
}
```

On the produce path, Confluent Gateway encrypts the tagged fields or the full payload
before writing to Kafka. On the fetch path, it decrypts them for consumers.

To keep client-side encryption while Confluent Gateway validates records, set
`enableAt` to `CLIENT`, and set `execution.environment` to `CLIENT` in
your client serializer configuration. If you don’t set `enableAt` and
Confluent Gateway has rule executors configured, both Confluent Gateway and client-side
serializers run the rules.

## Verify governance

To verify that Confluent Gateway enforces governance, register a schema for a topic,
enable governance for values, and produce records through the Confluent Gateway route
endpoint.

Before you start, set the `SR_URL`, `SR_USER`, `SR_PASSWORD`, and
`GATEWAY_BOOTSTRAP` environment variables, and create a `client.properties`
file with the security settings that your clients use for the Confluent Gateway route.
For more information, see [Reconfigure clients](co-gateway-deploy.md#co-gateway-reconfigure-clients).

1. Register an Apache Avro® schema for the `orders` topic value:
   ```bash
   curl -s -X POST \
     -H "Content-Type: application/vnd.schemaregistry.v1+json" \
     -u "$SR_USER:$SR_PASSWORD" \
     "$SR_URL/subjects/orders-value/versions" \
     --data '{"schemaType":"AVRO","schema":"{\"type\":\"record\",\"name\":\"Order\",\"namespace\":\"com.example\",\"fields\":[{\"name\":\"order_id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"currency\",\"type\":\"string\"}]}"}'
   ```
2. Enable governance for values in the Confluent Gateway CR:
   ```yaml
   spec:
     policies:
       dataGovernance:
         schemaRegistry: { ... }
         key:
           enabled: false
         value:
           enabled: true
   ```
3. Produce a record with the Schema Registry serializer:
   ```bash
   kafka-avro-console-producer \
     --bootstrap-server "$GATEWAY_BOOTSTRAP" \
     --topic orders \
     --producer.config client.properties \
     --property schema.registry.url="$SR_URL" \
     --property basic.auth.credentials.source=USER_INFO \
     --property basic.auth.user.info="$SR_USER:$SR_PASSWORD" \
     --property value.schema.id=<orders-value-schema-id>
   > {"order_id":"A-1001","amount":42.5,"currency":"USD"}
   ```

   Confluent Gateway accepts the record.
4. Produce a plain-text record, which has no schema ID:
   ```bash
   kafka-console-producer \
     --bootstrap-server "$GATEWAY_BOOTSTRAP" \
     --producer.config client.properties \
     --topic orders
   > {"order_id":"A-1002","amount":10.0,"currency":"USD"}
   ```

   Confluent Gateway rejects the record, and the producer receives an `INVALID_RECORD`
   error.

For more information about the errors that Confluent Gateway returns, see [Error
handling and response codes](https://docs.confluent.io/private-cloud-gateway/current/governance/gateway-governance-config.html#error-handling-and-response-codes)
for Confluent Private Cloud Gateway or [Error handling and response codes](https://docs.confluent.io/platform/current/gateway/governance/gateway-governance-config.html#error-handling-and-response-codes)
for the Confluent Gateway add-on for Confluent Platform.