Envelope Encryption is a technique described by NIST Special Publication 800-57 Part 1 Revision 5. It is the practice of encrypting the data with a Data Encryption Key (DEK), then wrapping (encrypting) the DEK with a Key Encryption Key (KEK). In this section we discuss how Envelope Encryption is employed by the filter to encrypt Kafka records. For more detail, refer to the design.

On the produce path, the filter is encrypting records. The filter uses a selector to determine which KEK to apply. It then requests that the KMS generates a DEK for the KEK. It is the DEK that is used to encrypt the record. The original record in the produce request is replaced by a cipher record which comprises the encrypted record, the encrypted DEK and some other metadata.

On the fetch path, the filter is decrypting records. The filter receives the cipher record from the Kafka Broker. The filter reverses the process used to construct the cipher record and uses the KMS to decrypt the DEK. The decrypted DEK is used to decrypt the encrypted record. The cipher record in the fetch response is then replaced with the decrypted record.

On the produce path, the filter employs a DEK reuse strategy. This means that records sent by a single connection to the same topic will be encrypted using the same DEK (until a time-out or encryption operations limit is reached, whichever occurs first). This prevents key exhaustion and prevents excessive interactions with the KMS.

On the fetch path, the filter employs an LRU strategy to keep recently encountered DEKs decrypted in memory. This prevents excessive interactions with the KMS.

The Key Management Service (KMS) is the secure repository for Key Encryption Keys (KEKs). The key material of the KEK never leaves the confines of the KMS. The KMS exposes cryptographic functions that operate against the KEKs stored within it.

The KMS is also the source of Data Encryption Keys (DEKs). When the KMS generates a DEK for a given KEK, it returns the DEK (which is securely generated random data), together with the encrypted DEK (which is the same data, encrypted using the KEK). Note that the KMS does not store the DEKs - DEKs are stored encrypted as part of the Kafka record held by the broker.

The filter uses the services of the KMS to generate DEKs, and decrypt encrypted DEKs.

The KMS must be available at runtime. If the KMS is unavailable, it will become impossible to produce or consume records through the filter until the KMS service is restored.

The filter currently has a single KMS integration with HashiCorp Vault®. More KMS integrations are planned. It is also possible for a user to implement their own KMS integration. The implementation must implement the KMS public API and make the implementation available on the classpath with the service loader mechanism.

Key rotation is the practice of periodically replacing cryptographic keys with new ones. Using key rotation is considered cryptographic best-practice.

The filter allows for the rotation of KEKs within the KMS. When a KEK is rotated, the new key material will be applied to newly produced records. Existing records (which were encrypted with older versions of the KEK) remain decryptable as long as the previous KEK version remains present in the KMS.

The filter currently encrypts only record values. Record keys, headers, timestamps are not encrypted.

Null record values (which represent deletes, or tombstones, in compacted topics) sent by producers are passed through to the broker unencrypted. This means encryption may be applied to compacted topics too. Deletes will function normally.

The presence of unencrypted records on a topic configured for encryption is allowed. The unencrypted records will get passed through to consumer as normal. This supports the use-case where encryption is introduced into a system where topics are already populated with unencrypted content.

The filter supports use-cases where some Kafka topics are configured for encryption while others are left to be unencrypted.

Transactions producing to both encrypted and unencrypted topics are supported.

In order to set up the KMS provider ready to use with the filter, follow these KMS provider specific steps.

The filter integrates with the HashiCorp Vault Transit Engine. Vault does not enable the Transit Engine by default. It must be enabled before it can be used with the filter.

The Vault Transit Engine URL is required so the filter knows the location of the Transit Engine within the Vault instance.

The URL is formed from the concatenation of the Api Address (reported by Vault reported by during starts up) with the complete path to Transit Engine, including the name of the engine itself. If Namespacing is used on the Vault instance, the path needs to include the namespace(s). The URL will end with /transit unless the -path parameter was used when enabling the engine.

If namespacing is not in use, the URL will look like this:

If namespacing is in use, the path must include the namespaces. For example, if there is a parent namespace is a and a child namespace is b, the URL will look like this:

If the name of the Transit engine was changed (using the -path argument to the vault secrets enable transit command) the URL will look like this:

It is necessary to determine a naming convention for the KEKs within Vault. This will allow the keys used by the filter to be kept separate from any keys used by other systems. This document assumes that the naming convention will be to prefix the keys used by the filter with the word KEK_. If a different naming convention is used, adapt the instructions accordingly.

To use the filter, there must be an administrative actor established. This actor, which is likely to be a human, has the responsibility to create keys within Vault for use by the filter.

The Administrator must have permissions to log in to Vault and create keys beneath transit/keys/KEK_* in the Vault hierarchy.

The exact steps required to establish an administrative actor will depend on the way that Vault instance has been setup.

A minimal Vault policy required by the Administrator is as follows:

To use the filter, there must be a Vault identity established for the filter itself. This identity must have permissions to perform the operations needed for envelope encryption (generating and decrypting DEKs).

Create a Vault policy for the filter actor:

Create a Periodic (long-lived) Vault Token for the filter^[1]:

The token create command yields the token. The token value is required later when configuring the vault within the filter.

The token must be renewed before expiration. It is the responsibility of the Administrator to do this.

This can be done with a command like:

To test whether the Kroxylicious Vault Token and the policy are working correctly, a script can be used.

First, as an Administrator, create a KEK in the hierarchy at this path transit/keys/KEK_testkey.

The script should respond 'Ok'. If errors are reported check the policy/token configuration.

transit/keys/KEK_testkey can now be removed.

The filter is configured as part of the filter chain in the following way:

The role of the KEK selector is to map from the topic name to key name. The filter looks up the resulting key name in the KMS.

It is the role of the Administrator to create KEKs in the KMS that will be used to encrypt the records. This must be done using whatever management interface the KMS provides.

The names (or aliases) of the encryption keys must match the naming conventions established above. If the selector generates a key name that doesn’t exist within the KMS, records will be sent to the topic without encryption.

For example, if using the TemplateKekSelector with the template KEK_${topicName}, create a key for every topic that is to be encrypted with the key name matching the topic name, prefixed by the string KEK_.

To verify that records sent to topics are indeed being encrypted, use kafka-console-consumer to consume the records directly from the target Kafka Cluster. Verify that encrypted text is seen rather than whatever plain text that was sent by producer.

The record values seen will look something like this:

To intercept all Fetch Requests your class would implement FetchRequestFilter:

To intercept all Fetch Responses your class would implement FetchResponseFilter:

To intercept all Fetch Requests and all Fetch Responses your class would implement FetchRequestFilter and FetchResponseFilter:

Specific Message Filter interfaces are mutually exclusive with Request/Response. Kroxylicious will reject invalid combinations of interfaces.

The FilterContext is the factory for the FilterResult objects.

There are two convenience methods^[2] that simply allow a filter to forward a result to the next filter. We’ve already seen these in action above.

To access richer features, use the filter result builders context.requestFilterResultBuilder() and responseFilterResultBuilder().

Filter result builders allow you to:

The builder lets you combine legal behaviours together. For instance, to close the connection after forwarding a response to a client, a response filter could use:

The builders yield either a completed CompletionStage<FilterResult> which can be returned directly from the filter method, or bare FilterResult. The latter exists to support asynchronous programming styles allowing you to use your own Futures.

This is a common pattern, we want to inspect or modify a message. For example:

In some cases we may wish to not forward a request from the client to the Cluster. Instead, we want to intercept that request and generate a response message in a Kroxylicious Protocol Filter and send it towards the client. This is called a short-circuit response.

For example:

This will respond to all Create Topic requests with an error response without forwarding any of those requests to the Cluster.

Filters can make additional asynchronous requests to the Cluster. This is useful if the Filter needs additional information from the Cluster in order to know how to mutate the filtered request/response.

The Filter can make use of CompletionStage chaining features ([#thenApply() etc.) to organise for actions to be done once the asynchronous request completes. For example, it could chain an action that mutates the filtered request/response using the asynchronous response, and finally, chain an action to forward the request/response to the next filter.

The asynchronous request/response will be intercepted by Filters upstream of this Filter. Filters downstream of this Filter (and the Client) do not see the asynchronous response.

Let’s take a look at an example. We’ll send an asynchronous request towards the Cluster for topic metadata while handling a FetchRequest and use the response to mutate the FetchRequest before passing it to the next filter in the chain.

As you have read above, we need to know the API version we want our request to be encoded at. Your filter can discover what versions of an API the Kafka Cluster supports. To do this use the ApiVersionsService available from the FilterContext to determine programmatically what versions of an API are support and then write code to make a suitable request object.

You may wish to restrict your Filter to only apply to specific versions of an API. For example, "intercept all FetchRequest messages greater than api version 7". To do this you can override a method named shouldHandleXyz[Request|Response] on your filter like:

In the PortPerBroker scheme, Kroxylicious automatically opens a port for each virtual cluster’s bootstrap and one port per broker of each target cluster. So, if you have two virtual clusters, each targeting a Kafka Cluster of three brokers, Kroxylicious will bind eight ports in total.

PortPerBroker is designed to work best with simplistic configurations. It is preferable if the target cluster has sequential, stable broker ids and a known minimum broker id (like 0,1,2 for a cluster of 3 brokers). It can work with non-sequential broker ids, but you would have to expose maxBrokerId - minBrokerId ports, which could be a huge number if your cluster included broker ids 0 and 20000.

Kroxylicious monitors the broker topology of the target cluster at runtime. It will adjust the number of open ports dynamically. If another broker is added to the Kafka Cluster, Kroxylicious will automatically open a port for it. Similarly, if a broker is removed from the Kafka Cluster, Kroxylicious will automatically close the port that was assigned to it.

The PortPerBroker scheme can be used with either clear text or TLS downstream connections.

The brokerAddressPattern configuration parameter understands the replacement token $(nodeId). It is optional. If present, it will be replaced by the node.id (or broker.id) assigned to the broker of the target cluster.

For example if your configuration looks like the above and your cluster has three brokers, your Kafka Client will receive broker address information like this:

The numberOfBrokerPorts imposes a maximum on the number of brokers that a Kafka Cluster can have. Set this value to be as high as the maximum number of brokers that your operational rules allow for a Kafka Cluster.

Note that each broker’s id must be greater than or equal to lowestTargetBrokerId, and less than lowestTargetBrokerId + numberOfBrokerPorts. The current strategy for mapping node ids to ports is nodeId → brokerStartPort + nodeId - lowestTargetBrokerId. So a configuration like:

can only map broker ids 1000, 1001 and 1002 to ports 9193, 9194 and 9195 respectively. You would have to reconfigure numberOfBrokerPorts to accommodate new brokers being added to the cluster.

In the SniRouting scheme, Kroxylicious uses SNI information to route traffic to either the boostrap or individual brokers. As this relies on SNI (Server Name Indication), the use of Downstream TLS is required.

With this scheme, you have the choice to share a single port across all virtual clusters, or you can assign a different port to each. Single port operation may have cost advantages when using load balancers of public clouds, as it allows a single cloud provider load balancer to be shared across all virtual clusters.

The brokerAddressPattern configuration parameters requires that the $(nodeId) token is present within it. This is replaced by the node.id (or `broker.id) assigned to the broker of the target cluster. This allows Kroxylicious to generate separate routes for each broker.

Defaults to binding to 0.0.0.0:9190. If no endpoints are configured the admin http server is not created.

This configuration:

Prometheus is connected to the micrometer global registry so Filters can record metrics against it, and they will be available as part of the Prometheus scrape data.

If you executed curl localhost:9999/metrics you should see metrics like:

We can add common tags that will be added to all metrics. These will be available as labels from the Prometheus scrape.

To configure common tags use configuration:

Micrometer has a concept of MeterBinder:

By registering some standard binders shipped with micrometer you can expose metrics about the JVM and system which can observe JVM memory usage, garbage collection and other behaviour.

To configure multiple binders you can use configuration like:

And those named binders will be bound to the global meter registry

Filters can use the static methods of Metrics to register metrics with the global registry. Or use Metrics.globalRegistry to get a reference to the global registry. Metrics registered this way will be automatically available through the prometheus scrape endpoint.