= Encryption Filter Design
Tom Bentley <tbentley@redhat.com>
:toc:
:icons: font
:source-highlighter: pygments

Design documentation on the record encryption filter using envelope encryption.

== Scope

The record encryption filter is designed to provide confidentiality and integrity guarantees for data stored in Apache Kafka(TM), under the threat models described later in this document.

Some assumptions are necessary in order to provide these guarantees, and these guarantees are not absolute. In other words, depending on how the proxy and filter are configured and deployed, it's possible that the actual security provided is less than what is described in this document. We will try to call-out in this document where we're making assumptions that affect the quality of security provided.

== Threat models

The aim is to provide additional security for data at-rest, from a threat actor who has operating system level access to the machines that comprise the Kafka cluster (i.e. brokers and controllers). We assume this actor can:

* Read files on disk
* Write files on disk
* Obtain a core dump of broker and controller process memory at any time
* Alter process memory at any time
* Observe TCP packets
* Craft TCP packets
* Kill and restart processes

We further assume that the threat actor

* has network access to the client side of the Kafka proxy, and/or
* has non-subverted access to existing Kafka client applications
// What does the above mean, exactly?

We assume the thread actor cannot interfere with proxy processes:

* Reconfigure it (to remove)
* Obtain a core dump of proxy process memory at any time

== Key management

=== Definitions

|===
|Term | Definition
|KMS [[kmsDef]] | Key Management System
|KEK [[kekDef]] | Key Encryption Key, a secret key that resides within the boundaries of a KMS
|DEK [[dekDef]] | Data Encryption Key, a secret key used to encrypt the data we want to protect at rest, could be generated by a KMS or the proxy
|EDEK [[edekDef]] | Encrypted Data Encryption Key, a DEK that has been encrypted using a KEK within the KMS boundary and transmitted to the proxy
|===

=== Assumptions

We assume the existence of a <<kmsDef, KMS>> which has not been subverted by the threat actor. Specifically:

1. all communication between the proxy and KMS is protected TLS
2. that the KMS never leaks a KEK
3. that the KMS never loses/deletes a KEK
4. that the KMS is always available

If assumption 1 is violated then all data encrypted with the exposed DEK is exposed (loss of confidentiality).

If assumption 2 is violated then all data encrypted by DEKs wrapped with those KEKs is exposed (loss of confidentiality).

If assumption 3 is violated then all data encrypted by DEKs wrapped with those KEKs becomes unreadable (loss of availability).

If assumption 4 is violated then all data encrypted by all DEKSs becomes unreadable (loss of availability), for the duration that the KMS is down – possibly forever.

When we say "data is exposed" we mean "becomes readable and modifiable by the threat actor, without detection".

=== Envelope Encryption

The filter uses Envelope Encryption.
The keys managed by the KMS are called Key Encryption Keys, (<<kekDef,KEKs>>).
Each KEK has a stable identifier which is used outside the KMS to refer to the KEK material, which never leaves the KMS.

The record data is encrypted by a data encryption key (<<dekDef,DEK>>), which is held (in plaintext) in-memory in the proxy process.

There are two ways a DEK might be generated, depending on the capabilities of the KMS used:

. The DEK is generated within the KMS, and transferred to the proxy over the network using TLS.
. The DEK is generated by the proxy process and the KMS is used to encrypt and later decrypt the DEK.

In either case, the DEK is encrypted (in the KMS) using the KEK and this DEK ciphertext (known as the <<edekDef,EDEK>>, or encrypted DEK) is stored with the encrypted record in the log on the (subverted) Kafka broker. Importantly, the secret KEK used to encrypt the DEK is _never_ known to the proxy.

When a record needs to be encrypted, if the DEK plaintext is not already known to the proxy is takes the encrypted DEK that's stored with the encrypted record and asks the KMS to decrypt it, then using the DEK plaintext is decrypted the encrypted record. The DEK plaintext may be kept in the proxy process memory, for a limited time, for use decrypting subsequent records.

=== KEK provisioning

The encryption filter does not support provisioning its own KEKs using the KMS. Instead, we assume that all KEKs are provisioned out of band.

=== KEK selection and encryption policy

KEK selection refers to how the filter determines which KEK to use for encrypting which records.
The filter only supports this decision being based on the record (not, for example, based on the identity of the Kafka client), and without
The filter only supports using a single key for encrypting record data.

// === KEK rotation
//
// TODO
//
// === DEK generation
//
// TODO Talk about when and how new DEKs get generated
//
// === DEK rotation
//
// TODO


== Encryption

=== Supported ciphers

Currently, the encryption filter supports

* AES-GCM with 256-bit key, 96 bit initialization vector and 128 bit authentication tag.

=== What to encrypt: The parcel

We use the term _parcel_ to refer to the collection of data that gets encrypted.
In other words, the parcel is the plaintext in the encryption operation.
In version v1 we only support encrypting:

* the record value
* optionally, the record headers

The following schema is used:
[source,abnf]
.The v1 parcel schema
----
parcel-v1              = record-value
                         num-headers
                         *header
num-headers            = 1*OCTET ; unsigned VARINT
header                 = name-length
                         name
                         value-length
                         value
name-length            = 1*OCTET ; unsigned VARINT
name                   = *OCTET
value-length           = 1*OCTET ; signed VARINT; -1 meaning null
value                  = *OCTET
record-value-length    = 1*OCTET ; signed VARINT; -1 meaning null
record-value           = *OCTET
----

NOTE: `VARINT` refers to the same varint encoding that it used in the Kafka protocol. It is a variable length encoding of a 32-bit integer. Small integers are encoded as a single byte. In the worst case 5 byes are required for encoding.

=== What to authenticate: The Additional Authenticated Data (AAD)

Using AAD makes it harder (but not impossible) for a threat actor with write access to manipulate log segments without detection.

NOTE: The Kafka record format prevents constructing an AAD schema that prevents all possible modifications to the log by a threat actor. For example, it is not practically possible to detect deletion of records from the log: For compacted logs this appears the same as legitimate record deletion by the log cleaner.

There are currently two options for the use of Additional Authenticated Data (AAD):

* None: No AAD
* Batch: Associates some of the metadata of the batch within which the record is contained
+
[source,abnf]
.v1 AAD schema
----
aad-batch              = producer-id
                         producer-epoch
                         producer-sequence-low
                         producer-sequence-high
                         record-key-length
                         record-key
producer-id            = 8OCTET ; <1>
producer-epoch         = 2OCTET ; <2>
producer-sequence-low  = 4OCTET ; <3>
producer-sequence-high = 4OCTET ; <4>
record-key-length      = 4OCTET ;
record-key             = *OCTET ;
----
<1> Taken from the `producerId` of the batch metadata in the Produce request
<2> Taken from the `producerEpoch` of the batch metadata in the Produce request
<3> Taken from the `baseSequence` of the batch metadata in the Produce request
<4> Computed from `baseSequence` + initial offset within the batch
+
The number of octets for each terminal are dictated by the Kafka message format. See <https://kafka.apache.org/documentation/#messageformat>

== Constructing the encrypted record

The "encrypted record" is the record which actually gets appended to the log.

First let's define the _wrapping_:

[source,abnf]
.v1 wrapping schema
----
wrapping-v1                = edek-length
                             edek
                             aad-code
                             cipher-code
                             cipher-specific-data
                             parcel-ciphertext-length
                             parcel-ciphertext
edek-length                = 1*OCTET      ; unsigned VARINT
edek                       = *OCTET       ; edek_length bytes
aad-code                   = OCTET        ; <1>
cipher-code                = OCTET        ; <2>
cipher-specific-data       = *OCTET       ; <3>
iv                         = 12OCTET      ; length implied by v1
parcel-ciphertext-length   = *OCTET ; VARINT
parcel-ciphertext          = *OCTET
----
<1> 0 means None, 1 means Batch (see AAD section)
<2> 0 means AES-GCM with 96 bit IV and 128 bit authentication tag
<3> The schema depends on the cipher. For `cipher-code=1` this is just the 12 byte IV.

For reference, this is the schema of a Kafka record:

[source,abnf]
.Kafka's record schema
----
record                     = length
                             attributes
                             timestampDelta
                             offsetDelta
                             recordKeyLength
                             recordKey
                             recordValueLen
                             recordValue
                             *recordHeader
recordHeader               = headerKeyLength
                             headerKey
                             headerValueLength
                             headerValue
----

To construct the "ciphertext record":

. Construct the `parcel` from the plaintext record's `value` and optionally the `headers`
. Encrypt the `parcel`, and construct the `wrapping`
. Create a copy of the plaintext record, but:
.. replace the `value` with the `wrapping` (adjusting the `valueLen`)
.. optionally clear the `headers` (if they were included in the parcel)
.. Prepend a header with name "kroxylicious.io/encryption" and byte value `0x01`.


=== Compression

Kafka supports compression of record batches (not individual records). This means there are a number of places where compression could be used:

==== Client sends a compressed batch
It's the _batches_ that are compressed (by the producer), but it is the _records_ that are to be encrypted, so we have no choice but to decompress the whole batch if we're to encrypt any of the records within it.
// todo: clarify whether this is talking about compressing individual records or batches.
There is the possibility of (re-)compressing records prior to encryption.
Compress-then-encrypt has been used for cryptanalysis (e.g. the CRIME vulnerability in HTTP). So we don't currently do this.


==== Proxy compresses a batch of encrypted records, prior to forwarding to a broker

It's the proxy's choice about which compression to use. In other words whatever compressor is used a proxied client won't be exposed to the compressed batches.

Since compression is a trade-off between CPU and compression ratio, we offer compression as a configurable option.

==== Proxy decompresses a batch of records, when receiving from a broker

The proxy is stateless. In particular the proxy does not maintain state about which topics have had encrypted records produced to them. This means there is no state to manage, and users can easily change which records are encrypted, and how, with confidence that existing records will still be readable.

However, as consequence of this design, on the fetch path we have no choice but to decompress batches in order to determine whether they contain encrypted records. Encrypted records are identified by the presence of the "kroxylicious.io/encryption" header.

==== Proxy compresses a batch of decrypted records, prior to forwarding to the client

The benefits of doing this are unclear. The compression only benefits a single client and only in terms of network bandwidth. It's unclear whether it's worth the CPU. Likely we should only support a single compressor, chosen based on experiment.

// == Kafka protocol considerations
//
// === Errors
//
// TODO Describe how problems during encryption/decryption map to Kafka error codes
//
// == Performance considerations
//
// TODO describe the threading model and main perf-related tunables
// TODO provide some indicative numbers for the effect on produce/fetch throughput and latency (including tail latencies)
//
// == Operational considerations
//
// TODO describe the metrics/monitoring, memory usage considerations etc.