# Field encryption

> Available since 1.6.0. The fields a message declares sensitive are encrypted by the gateway on write and decrypted for the one scope allowed to: the broker…

Guide - Since version 1.6.0 - Flexway v1.6.3 - https://flexway.io/en/docs/operations-kafka-encryption/

> Operations guide: configure, deploy and observe the gateway. What the capability does, product side: [product guide](https://flexway.io/en/docs/product-overview/); calling it: [developer guide](https://flexway.io/en/docs/developer-overview/).

Available since `1.6.0`. The fields a
message declares sensitive are encrypted **by the gateway** on write and decrypted for the one scope
allowed to: the broker, its backups and any reader without that scope never see the clear value. The
rest of the message stays readable downstream.

**How.** Envelope encryption: the gateway draws an AES-256 data key (`crypto/rand`), encrypts each field
with AES-256-GCM under a random 96-bit IV, and has the KMS **wrap** the data key under a key that never
leaves it. The wrapped key travels in the record's `flexway-dek` header. Each field is bound by its
additional data to the **tenant**, the **physical topic**, the **field path**, the version and the epoch:
a field copied to another field, topic or tenant does not decrypt. An encrypted field is written as
base64 in a text field, as bytes in a `bytes` field. In Avro and Protobuf, a path goes through records
and messages, never a list or a map, and ends on text or bytes. The `flexway-dek` header **also names
the fields it encrypted**, under an authentication code (HMAC-SHA256 under a key derived from the data
key): a reader decrypts what the record says it holds, not what the contract declares today. A path
removed from `x-flexway-encrypt` still decrypts in the history (as long as the schema declares the field
with the same type); a path added never makes the history unreadable; changing the list makes a new data
key. Removing `x-flexway-encrypt` altogether stops decryption: the history is handed on encrypted, with
its header.

**The KMS** - one kind per gateway, credentials always by reference or mounted file:

```yaml
# flexway.yaml - Vault transit, Kubernetes authentication (the pod's service account)
kms:
  key: flexway-kafka                   # the transit key; flexway-kafka-{tenant}: one key per tenant
  vault_transit:
    address: https://vault.security:8200   # https required: the token never travels in clear
    mount: transit                     # default
    kubernetes: { role: flexway }      # or token: ${VAULT_TOKEN} / file:/vault/token - never the value
```

The Vault policy of the gateway's role: `update` on encrypt and decrypt, `read` on the keys' metadata - **never
`create`**. With `create`, Vault transit makes a missing key at the first encrypt: a destroyed key would come
back, empty, under the same name - new writes go through under it, and the erased records, which no longer
authenticate, block their consumers (`CORRUPT_MESSAGE`) instead of being handed on with `key_destroyed`.

```hcl
# Vault policy of the gateway (flexway-kafka-{tenant}: the wildcard covers every tenant's key)
path "transit/encrypt/flexway-kafka*" { capabilities = ["update"] }
path "transit/decrypt/flexway-kafka*" { capabilities = ["update"] }
path "transit/keys/flexway-kafka*"    { capabilities = ["read"] }
```

```yaml
# AWS KMS: the SDK's default chain (IRSA, instance role) without credentials_file
kms:
  key: alias/flexway-kafka
  aws_kms: { region: eu-west-3 }
```

```yaml
# Google Cloud KMS: default credentials (Workload Identity) without credentials_file
kms:
  key: projects/acme/locations/europe-west1/keyRings/flexway/cryptoKeys/kafka
  gcp_kms: {}
```

```yaml
# local: key files in a mounted directory - tests and development only
kms:
  key: dev
  local: { dir: /keys }
```

**The contract.** The message declares its encrypted fields, the read operation the scope that
decrypts:

```yaml
# api/asyncapi.yaml - excerpt
channels:
  orders:
    address: "{tenant}.orders"
    parameters: { tenant: { x-flexway-bind: tenant } }
    messages:
      order:
        payload: { $ref: "#/components/schemas/Order" }
        x-flexway-encrypt: [customer.email, payment.iban]   # dotted paths; no index nor wildcard
operations:
  readOrders:
    action: receive
    channel: { $ref: "#/channels/orders" }
    x-flexway-kafka-proxy: { cluster: orders, decrypt_scope: orders:decrypt }
```

A reader whose token carries `orders:decrypt` receives the fields in clear; others receive the encrypted
value. `decrypt_scope` exists only on a `receive` operation whose message declares `x-flexway-encrypt`:
otherwise the contract is refused at load. A `null` value is never encrypted.

**Settings** - each has a default and a bound, refused at load beyond it:

| Key | Default | Bounds | Why |
| --- | --- | --- | --- |
| `kms.timeout` | 2 s | up to 30 s | A KMS call holds the Kafka request that needs it |
| `kms.rotation.interval` | 1 h | 1 min to 7 d | New epoch, new data key: an erasure or a leak covers a bounded window |
| `kms.rotation.max_encryptions` | 5,000,000 | 1 to 2^30 | AES-GCM with random IVs stays far below 2^32 fields per key |
| `kms.previous_keys` | none | 8 at most, each like `kms.key` | The keys a record may still name - `kms.key`'s former values, read, never written. A header naming any other key is refused (`CORRUPT_MESSAGE`, 2) without a KMS call |
| `kms.cache.max_keys` | 10,000 | 1 to 1,000,000 | Data keys kept in clear in memory, wiped on eviction; past it, the oldest key **of the tenant holding the most** goes: a tenant replaying its history never evicts the others' keys |
| `kms.cache.ttl` | 1 h | 1 s to 24 h | An unwrapped data key's life **from its unwrap**, used or not; and how often the current write key asks the KMS whether its key still holds. It is the longest an erasure waits (below) |
| `kms.cache.negative_ttl` | 5 s | 100 ms to 5 min | A KMS refusal remembered: no call storm |
| `kms.tenant_calls.rate` / `burst` | 10 / 20 | rate > 0, burst >= 1 | KMS calls per tenant per second: one tenant never exhausts the KMS of the others |

**KMS unavailable, key destroyed** - the
gateway **never** publishes in clear a field declared encrypted and **never** silently hands the
ciphertext to a reader holding the scope:

| Situation | Producer | Consumer with the scope | What to do |
| --- | --- | --- | --- |
| KMS unreachable, data key still cached | Served normally | Served while the read epoch's key is cached | Restore the KMS; rule `kms-unavailable` firing |
| KMS unreachable, no valid key left | `KAFKA_STORAGE_ERROR` (56), retriable: replayed within `delivery.timeout.ms`, nothing published | `KAFKA_STORAGE_ERROR` (56) on the partition: it fetches again | Restore the KMS before `delivery.timeout.ms` |
| KMS access refused (Vault policy, IAM) | Like an outage (56) | Like an outage (56) | Fix the policy |
| Key destroyed, disabled or version retired: crypto-shredding | `RESOURCE_NOT_FOUND` (91), not retriable; a version retired alone: a new data key under the latest version, the write goes through | Record **handed on**, field **still encrypted**, header `flexway-encryption: key_destroyed` - it **must** read that header before using the field | Wanted: the fields are unreadable for good; otherwise restore the key version |
| A field or wrapped key that does not authenticate (tampered) | - | `CORRUPT_MESSAGE` (2) on the partition, never a value | Find the writer that bypasses the gateway |

```python
# Consumer with the scope (confluent-kafka, Python): an erased field stays encrypted
for msg in consumer.consume(100, timeout=1.0):
    headers = dict(msg.headers() or [])
    if headers.get("flexway-encryption") == b"key_destroyed":
        continue  # wanted erasure: do not use the encrypted fields
    handle(msg.value())
```

**What each KMS calls "destroyed"** - a key is judged erased only on an explicit state; a transition, a
typo or an access refused is an outage (56), never an erasure:

| KMS answer | On write (wrap) | On read (unwrap) |
| --- | --- | --- |
| Vault: `encryption key not found` | Outage (56): `kms.key` names no key | Erasure (`key_destroyed`) |
| Vault: a version below `min_decryption_version` (`disallowed by policy (too old)`) | - (a write takes the latest version) | Erasure; the write key it wrapped is made again |
| Vault: `invalid ciphertext`, `message authentication failed` (a forged `vault:v999` included) | - | Invalid (`CORRUPT_MESSAGE`, 2) |
| AWS: `DisabledException`; `KMSInvalidStateException` of a key `Disabled` or `PendingDeletion` (read by `DescribeKey`) | Erasure (91) | Erasure |
| AWS: `NotFoundException` | Outage (56) | Erasure (the key wrapped this data) |
| AWS: `KMSInvalidStateException` of a key `Creating`, `Updating`, `PendingImport`, `Unavailable`, or a state not read | Outage (56) | Outage (56) |
| GCP: `FAILED_PRECONDITION` naming the state `DISABLED`, `DESTROYED` or `DESTROY_SCHEDULED` | Erasure (91) | Erasure |
| GCP: `NOT_FOUND` | Outage (56) | Erasure |
| GCP: any other `FAILED_PRECONDITION` (external key manager unreachable, import in progress) | Outage (56) | Outage (56) |
| All: access refused, timeout, `5xx`, network | Outage (56) | Outage (56) |

A key pending deletion (AWS) or destruction (GCP) is **disabled** - the ADR's erasure; cancelling the
deletion makes it readable again, but the records already handed on marked `key_destroyed` do not come
back.

**How long an erasure takes.** A key destroyed in the KMS takes effect on every replica within
**`kms.cache.ttl` at most** (1 h by default), plus `kms.cache.negative_ttl`: an unwrapped data key lives
`kms.cache.ttl` from its unwrap, used or not, and the write key asks the KMS again every
`kms.cache.ttl` (an unwrap of its own wrapped key, no extra permission). If the KMS is unreachable then,
the write key goes on (T2) until `kms.rotation.interval` and the KMS is asked again every
`kms.cache.negative_ttl`. A shorter `kms.cache.ttl` shortens the delay, at the cost of more KMS calls. A
replica started after a deleted (not found) key's erasure refuses writes as an outage (56) - the KMS does
not tell a deleted key from one never created.

**Key check at start.** At start and at each reload that changes the KMS, the gateway reads `kms.key`'s
metadata: an absent key, a Vault transit key not derived (`derived=true` required), a non-symmetric AWS
key or a GCP key of another purpose **refuse** the configuration; a disabled key is taken (writes get
91); a KMS unreachable or a permission missing block nothing and are logged. Read permission to grant:
Vault `read` on `<mount>/keys/<key>`, AWS `kms:DescribeKey`, GCP `cloudkms.cryptoKeys.get`
(`roles/cloudkms.viewer`). A name with `{tenant}` is not checked at start: a tenant's missing key refuses
its writes as an outage (56).

**Erasing a tenant's data.** With one key per tenant (`key: flexway-kafka-{tenant}`), destroying its key
in the KMS (Vault: `deletion_allowed` then delete; AWS: `ScheduleKeyDeletion`; GCP: destroy the versions)
makes its fields unreadable for good, the other tenants' intact. With a shared key, retiring a version
(Vault's `min_decryption_version`) erases the epochs it wrapped. The console counts these reads apart
(`key_destroyed`), never as an outage.

**What is refused.** A record produced with a `flexway-*` header (`INVALID_RECORD`, 87): those headers
belong to the gateway. A record whose schema does not decode a declared field, or that holds a declared
field twice (a repeated JSON key) (`INVALID_RECORD`, 87). A control batch (bit `0x20`) in a `Produce`, on
**any** served topic, encrypted or not (`INVALID_RECORD`, 87): only the transaction coordinator writes
one. A contract declaring `x-flexway-encrypt` without `kms` in `flexway.yaml`, or with
`listeners.kafka.max_response_bytes` below `max_request_bytes` (a record written could not be read back).

**Accepted limits.** A record's fields are bound to the tenant, the topic, the path and the epoch, not to
the record: whoever writes to the broker without the gateway can swap two fields encrypted under the same
data key between two records of the same topic; they can neither read them nor move them to another
field, topic or tenant. In Protobuf, a field present twice on the wire keeps one - the last, Protobuf's
merge rule - rewritten encrypted: no clear copy goes through.

**Signals.** `flexway.kafka.encryption{cluster, tenant, op, outcome}` (`ok`, `kms_unavailable`,
`key_destroyed`, `invalid`), `flexway.kafka.kms.calls{cluster, op, outcome}` (`wrap`, `unwrap`;
`budget_exhausted` when the tenant's budget refuses without a call), `flexway.kafka.dek.cache`,
`flexway.kafka.dek.epoch` and `flexway.kafka.dek.cached`. `/config` publishes the kind, the host, the key
name, the rotation and the cache, **never** a reference, a secret or a key. In the console, the cluster's
panel (Upstreams) shows the KMS and its counters, and the rules `kms-unavailable` and
`kafka-key-destroyed` (disabled by default) alert ([Console: virtual clusters, Kafka clients, rules](https://flexway.io/en/docs/operations-kafka-clients/#console-virtual-clusters-kafka-clients-rules)).

| Symptom | Cause | What to do |
| --- | --- | --- |
| Start refused, `kms: write exactly one of ...` | No KMS kind or several | Write exactly one |
| Start refused, `kms.vault_transit.token must be a reference` | The token written in clear | `${VAULT_TOKEN}` or `file:/path` |
| Producers get `KAFKA_STORAGE_ERROR` | KMS unreachable or access refused, cache exhausted | Restore the KMS or its policy; `kms.cache.ttl` sets the autonomy |
| A reader with the scope sees encrypted values | Header `key_destroyed`: key erased | Wanted; otherwise restore the key version |
| `INVALID_RECORD` on produce | A client's `flexway-*` header, a value its schema does not decode, a declared field written twice, or a control batch | Drop the header; fix the value; never mark a data batch as a control batch |
| Start refused, `kms.key "...": kms key absent or unusable` | The key does not exist (a typo), a Vault key not derived, a non-symmetric AWS key, a GCP key of another purpose | Fix `kms.key`, or create the key (`derived=true` for Vault) |
| Log `kms.key could not be checked` | KMS unreachable, or the permission to read its metadata not granted | Grant `read` (Vault), `kms:DescribeKey` (AWS), `cloudkms.cryptoKeys.get` (GCP) - optional |
| `CORRUPT_MESSAGE` on a partition after `kms.key` changed | The records name the former key | Declare it in `kms.previous_keys` |
| `CORRUPT_MESSAGE` after a schema change | A field the history encrypts no longer exists in the schema, or changed type | Keep the field in the schema while the history holds it |