de.saly.kafka.crypto

Class EncryptingSerializer<T>

java.lang.Object

de.saly.kafka.crypto.SerdeCryptoBase

de.saly.kafka.crypto.EncryptingSerializer<T>

Type Parameters:

T - The type to be serialized from (applied to the wrapped serializer)

All Implemented Interfaces:

org.apache.kafka.common.serialization.Serializer<T>

public class EncryptingSerializer<T>
extends SerdeCryptoBase
implements org.apache.kafka.common.serialization.Serializer<T>

This is a serialization wrapper which adds message encryption. Its intended to be used together with DecryptingDeserializer

Configuration

• crypto.rsa.publickey.filepath path on the local filesystem which hold the RSA public key (X.509 format) of the consumer
• crypto.wrapped_serializer is the class or full qualified class name or the wrapped serializer
• crypto.hash_method Type of hash generated for the AES key (optional, default is "SHA-256")
• crypto.new_key_msg_interval Generate new AES every n messages (default is -1, that means never generate a new key)

Each message is encrypted with "AES/CBC/PKCS5Padding" before its sent to Kafka. The AES key as well as the initialization vector are random. The AES key is attached to the message in a RSA encrypted manner. The IV is also attached but not RSA encrypted. There is also a hash value of the AES key to allow consumers caching of decrypted AES keys. Finally we have a few magic and header bytes. The resulting byte array looks therefore like this:

MMLLLHH..HHEEEE..EEEEIIII..IIIOOOOO....OOOOOO

• MM: Two magic bytes 0xDF 0xBB to detect if this byte sequence is encrypted or not
• LLL: Three bytes indicating the length of the AES key hash, the RSA encrypted AES key and the IV
• HH..HH: AES key hash
• EE..EE: RSA encrypted AES key
• II..II: Initialization vector (if any)
• OO..OO: The AES encrypted original message

MMLLL is called the encryption header and consists of 5 bytes.

• M1: 0xDF
• M2: 0xBB
• L1: length of the AES key hash
• L2: RSA factor f so that f*128*8 evaluates to the RSA keysize (in bits)
• L3: length of the initialization vector in bytes (always 16 for AES CBC)

RSA public/private keypair can be generated with
java -cp kafka-end-2-end-encryption-1.0.0.jar de.saly.kafka.crypto.RsaKeyGen 2048

Note: As Producers are multithreading-safe this serializer is also thread-safe

Field Summary

Fields

Modifier and Type	Field and Description
static String	CRYPTO_NEW_KEY_MSG_INTERVAL
static String	CRYPTO_VALUE_SERIALIZER
int	msgInterval

Fields inherited from class de.saly.kafka.crypto.SerdeCryptoBase

CRYPTO_AES_KEY_LEN, CRYPTO_HASH_METHOD, CRYPTO_IGNORE_DECRYPT_FAILURES, CRYPTO_RSA_PRIVATEKEY_FILEPATH, CRYPTO_RSA_PUBLICKEY_FILEPATH, DEFAULT_TRANSFORMATION

Constructor Summary

Constructors

Constructor and Description
EncryptingSerializer()

Method Summary

Modifier and Type	Method and Description
void	close()
void	configure(Map<String,?> configs, boolean isKey)
void	newKey() Generate new AES key for the current thread
byte[]	serialize(String topic, T data)

Methods inherited from class de.saly.kafka.crypto.SerdeCryptoBase

concatenate, crypt, init, newInstance

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

CRYPTO_VALUE_SERIALIZER

public static final String CRYPTO_VALUE_SERIALIZER

See Also:

Constant Field Values

CRYPTO_NEW_KEY_MSG_INTERVAL

public static final String CRYPTO_NEW_KEY_MSG_INTERVAL

See Also:

Constant Field Values

msgInterval

public int msgInterval

Constructor Detail

EncryptingSerializer

public EncryptingSerializer()

Method Detail

configure

public void configure(Map<String,?> configs,
                      boolean isKey)

Specified by:

configure in interface org.apache.kafka.common.serialization.Serializer<T>

serialize

public byte[] serialize(String topic,
                        T data)

Specified by:

serialize in interface org.apache.kafka.common.serialization.Serializer<T>

close

public void close()

Specified by:

close in interface org.apache.kafka.common.serialization.Serializer<T>

newKey

public void newKey()

Generate new AES key for the current thread

Overrides:

newKey in class SerdeCryptoBase