Field encryption

Operations guide: configure, deploy and observe the gateway. What the capability does, product side: product guide; calling it: developer guide.

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:

# 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.

# 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"] }

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

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

# 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:

# 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

# 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).

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