Encryption at Rest: Protecting Kafka Passive Data from Physical Theft #

# Initializing a LUKS encrypted partition on the /dev/sdb disk
# We'll be asked to enter a master passphrase to lock the disk
sudo cryptsetup luksFormat /dev/sdb

# Opening the encrypted disk and naming it "kafka_data_crypt"
sudo cryptsetup open /dev/sdb kafka_data_crypt

# Creating an ext4 file system
sudo mkfs.ext4 /dev/mapper/kafka_data_crypt

# Creating the mount point directory
sudo mkdir -p /var/lib/kafka/data

# Mounting the encrypted volume to the Kafka data directory
sudo mount /dev/mapper/kafka_data_crypt /var/lib/kafka/data

# Changing directory ownership so the kafka user can access it
sudo chown -R kafka:kafka /var/lib/kafka/data

# Creating a random key file
sudo dd if=/dev/urandom of=/etc/security/kafka-disk.key bs=1024 count=4
sudo chmod 400 /etc/security/kafka-disk.key

# Adding the key file to the disk's LUKS key slots
sudo cryptsetup luksAddKey /dev/sdb /etc/security/kafka-disk.key

kafka_data_crypt  /dev/sdb  /etc/security/kafka-disk.key  luks

/dev/mapper/kafka_data_crypt  /var/lib/kafka/data  ext4  defaults,noatime  0  2

package com.mycompany.kafka.security;

import org.apache.kafka.common.serialization.Serializer;
import java.nio.ByteBuffer;
import java.security.SecureRandom;
import java.util.Map;
import javax.crypto.Cipher;
import javax.crypto.spec.GCMParameterSpec;
import javax.crypto.spec.SecretKeySpec;

public class EnvelopeEncryptionSerializer implements Serializer<byte[]> {

    private String kmsKeyId;
    private MockKmsClient kmsClient;
    private SecureRandom secureRandom;

    // Secure local DEK cache to optimize performance
    private CachedDek currentCachedDek;
    private static final long DEK_ROTATION_INTERVAL_MS = 300000; // Rotate the DEK every 5 minutes

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        this.kmsKeyId = (String) configs.get("encryption.kms.key.id");
        // In production, replace with the real AWS KMS client / HashiCorp Vault client
        this.kmsClient = new MockKmsClient();
        this.secureRandom = new SecureRandom();
    }

    private synchronized CachedDek getOrGenerateDek() {
        long now = System.currentTimeMillis();
        if (currentCachedDek == null || (now - currentCachedDek.createdAtMs) > DEK_ROTATION_INTERVAL_MS) {
            // Request a new DEK from the KMS
            KmsDekEnvelope envelope = kmsClient.generateDataKey(kmsKeyId);
            currentCachedDek = new CachedDek(envelope.plaintextKey, envelope.encryptedKey, now);
        }
        return currentCachedDek;
    }

    @Override
    public byte[] serialize(String topic, byte[] data) {
        if (data == null) {
            return null;
        }

        try {
            // 1. Get the DEK (from the cache or create a new one via the KMS)
            CachedDek dek = getOrGenerateDek();

            // 2. Create a random IV (Initialization Vector) for AES-GCM (12 bytes)
            byte[] iv = new byte[12];
            secureRandom.nextBytes(iv);

            // 3. Do AES-GCM encryption on the data payload
            Cipher cipher = Cipher.getInstance("AES/GCM/NoPadding");
            SecretKeySpec keySpec = new SecretKeySpec(dek.plaintextKey, "AES");
            GCMParameterSpec parameterSpec = new GCMParameterSpec(128, iv); // 128-bit tag length
            cipher.init(Cipher.ENCRYPT_MODE, keySpec, parameterSpec);
            byte[] encryptedPayload = cipher.doFinal(data);

            // 4. Assemble the output message with the format:
            // [Encrypted DEK Length (4 bytes)] + [Encrypted DEK] + [IV (12 bytes)] + [Encrypted Payload]
            byte[] encryptedDekBytes = dek.encryptedKey;
            int totalLength = 4 + encryptedDekBytes.length + 12 + encryptedPayload.length;
            
            ByteBuffer buffer = ByteBuffer.allocate(totalLength);
            buffer.putInt(encryptedDekBytes.length);
            buffer.put(encryptedDekBytes);
            buffer.put(iv);
            buffer.put(encryptedPayload);

            return buffer.array();
        } catch (Exception e) {
            throw new RuntimeException("Failed to encrypt the Kafka payload: ", e);
        }
    }

    @Override
    public void close() {
        if (currentCachedDek != null) {
            currentCachedDek.wipe();
        }
    }

    // Helper class for holding the DEK currently in use
    private static class CachedDek {
        final byte[] plaintextKey;
        final byte[] encryptedKey;
        final long createdAtMs;

        CachedDek(byte[] plaintextKey, byte[] encryptedKey, long createdAtMs) {
            this.plaintextKey = plaintextKey;
            this.encryptedKey = encryptedKey;
            this.createdAtMs = createdAtMs;
        }

        void wipe() {
            java.util.Arrays.fill(plaintextKey, (byte) 0); // Wipe the plain key from memory
        }
    }
}

package com.mycompany.kafka.security;

import org.apache.kafka.common.serialization.Deserializer;
import java.nio.ByteBuffer;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import javax.crypto.Cipher;
import javax.crypto.spec.GCMParameterSpec;
import javax.crypto.spec.SecretKeySpec;

public class EnvelopeDecryptionDeserializer implements Deserializer<byte[]> {

    private MockKmsClient kmsClient;
    
    // Cache of Encrypted DEK -> Plaintext DEK to cut decryption latency
    private ConcurrentHashMap<ByteBuffer, byte[]> decryptedKeysCache;

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        this.kmsClient = new MockKmsClient();
        this.decryptedKeysCache = new ConcurrentHashMap<>();
    }

    private byte[] getOrDecryptDek(byte[] encryptedDek) {
        ByteBuffer wrapper = ByteBuffer.wrap(encryptedDek);
        if (decryptedKeysCache.containsKey(wrapper)) {
            return decryptedKeysCache.get(wrapper);
        }

        // Decrypt using the KMS if not in the cache
        byte[] plaintextKey = kmsClient.decryptDataKey(encryptedDek);
        decryptedKeysCache.put(wrapper, plaintextKey);
        return plaintextKey;
    }

    @Override
    public byte[] deserialize(String topic, byte[] data) {
        if (data == null) {
            return null;
        }

        try {
            ByteBuffer buffer = ByteBuffer.wrap(data);

            // 1. Extract the Encrypted DEK
            int encryptedDekLength = buffer.getInt();
            byte[] encryptedDek = new byte[encryptedDekLength];
            buffer.get(encryptedDek);

            // 2. Extract the IV
            byte[] iv = new byte[12];
            buffer.get(iv);

            // 3. Extract the Encrypted Payload
            int payloadLength = buffer.remaining();
            byte[] encryptedPayload = new byte[payloadLength];
            buffer.get(encryptedPayload);

            // 4. Get the Plaintext DEK
            byte[] plaintextKey = getOrDecryptDek(encryptedDek);

            // 5. Decrypt the data using AES-GCM
            Cipher cipher = Cipher.getInstance("AES/GCM/NoPadding");
            SecretKeySpec keySpec = new SecretKeySpec(plaintextKey, "AES");
            GCMParameterSpec parameterSpec = new GCMParameterSpec(128, iv);
            cipher.init(Cipher.DECRYPT_MODE, keySpec, parameterSpec);
            
            return cipher.doFinal(encryptedPayload);
        } catch (Exception e) {
            throw new RuntimeException("Failed to decrypt the Kafka payload: ", e);
        }
    }

    @Override
    public void close() {
        decryptedKeysCache.values().forEach(key -> java.util.Arrays.fill(key, (byte) 0));
        decryptedKeysCache.clear();
    }
}

package com.mycompany.kafka.security;

import java.security.SecureRandom;
import javax.crypto.Cipher;
import javax.crypto.spec.SecretKeySpec;

public class MockKmsClient {

    private final byte[] localMasterKek;
    private final SecureRandom secureRandom;

    public MockKmsClient() {
        this.secureRandom = new SecureRandom();
        this.localMasterKek = new byte[32]; // Simulate a 256-bit KEK
        secureRandom.nextBytes(localMasterKek);
    }

    public KmsDekEnvelope generateDataKey(String kmsKeyId) {
        // 1. Generate a Plaintext DEK (256-bit AES)
        byte[] plaintextKey = new byte[32];
        secureRandom.nextBytes(plaintextKey);

        // 2. Encrypt (Wrap) the DEK using localMasterKek (KEK)
        byte[] encryptedKey = wrapKey(plaintextKey);

        return new KmsDekEnvelope(plaintextKey, encryptedKey);
    }

    public byte[] decryptDataKey(byte[] encryptedKey) {
        // Decrypt (Unwrap) the DEK using localMasterKek (KEK)
        return unwrapKey(encryptedKey);
    }

    private byte[] wrapKey(byte[] plaintextKey) {
        try {
            Cipher cipher = Cipher.getInstance("AES");
            SecretKeySpec keySpec = new SecretKeySpec(localMasterKek, "AES");
            cipher.init(Cipher.WRAP_MODE, keySpec);
            return cipher.wrap(new SecretKeySpec(plaintextKey, "AES"));
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private byte[] unwrapKey(byte[] encryptedKey) {
        try {
            Cipher cipher = Cipher.getInstance("AES");
            SecretKeySpec keySpec = new SecretKeySpec(localMasterKek, "AES");
            cipher.init(Cipher.UNWRAP_MODE, keySpec);
            return cipher.unwrap(encryptedKey, "AES", Cipher.SECRET_KEY).getEncoded();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

class KmsDekEnvelope {
    final byte[] plaintextKey;
    final byte[] encryptedKey;

    KmsDekEnvelope(byte[] plaintextKey, byte[] encryptedKey) {
        this.plaintextKey = plaintextKey;
        this.encryptedKey = encryptedKey;
    }
}