Encryption at Rest: Melindungi Data Pasif Kafka dari Pencurian Fisik #

# Menginisialisasi partisi terenkripsi LUKS pada disk /dev/sdb
# Kita akan diminta memasukkan passphrase induk untuk mengunci disk
sudo cryptsetup luksFormat /dev/sdb

# Membuka disk terenkripsi dan menamainya "kafka_data_crypt"
sudo cryptsetup open /dev/sdb kafka_data_crypt

# Membuat sistem berkas ext4
sudo mkfs.ext4 /dev/mapper/kafka_data_crypt

# Membuat direktori mount point
sudo mkdir -p /var/lib/kafka/data

# Memasang volume terenkripsi ke direktori data Kafka
sudo mount /dev/mapper/kafka_data_crypt /var/lib/kafka/data

# Ubah hak kepemilikan direktori agar dapat diakses oleh user kafka
sudo chown -R kafka:kafka /var/lib/kafka/data

# Membuat berkas kunci acak
sudo dd if=/dev/urandom out=/etc/security/kafka-disk.key bs=1024 count=4
sudo chmod 400 /etc/security/kafka-disk.key

# Menambahkan berkas kunci ke slot kunci LUKS disk
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;

    // Cache DEK lokal yang aman untuk mengoptimalkan performa
    private CachedDek currentCachedDek;
    private static final long DEK_ROTATION_INTERVAL_MS = 300000; // Rotasi DEK setiap 5 menit

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        this.kmsKeyId = (String) configs.get("encryption.kms.key.id");
        // Di produksi, ganti dengan client AWS KMS asli / 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) {
            // Minta DEK baru dari 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. Dapatkan DEK (dari cache atau buat baru via KMS)
            CachedDek dek = getOrGenerateDek();

            // 2. Buat IV (Initialization Vector) acak untuk AES-GCM (12 bytes)
            byte[] iv = new byte[12];
            secureRandom.nextBytes(iv);

            // 3. Lakukan enkripsi AES-GCM terhadap payload data
            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. Susun pesan keluaran dengan format:
            // [Panjang Encrypted DEK (4 bytes)] + [Encrypted DEK] + [IV (12 bytes)] + [Payload Terenkripsi]
            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("Gagal mengenkripsi payload Kafka: ", e);
        }
    }

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

    // Kelas pembantu untuk menampung DEK yang sedang digunakan
    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); // Bersihkan kunci polos dari memori
        }
    }
}

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 DEK terenkripsi -> DEK Plaintext untuk memotong latensi dekripsi
    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);
        }

        // Dekripsi menggunakan KMS jika tidak ada di 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. Ekstrak Encrypted DEK
            int encryptedDekLength = buffer.getInt();
            byte[] encryptedDek = new byte[encryptedDekLength];
            buffer.get(encryptedDek);

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

            // 3. Ekstrak Payload Terenkripsi
            int payloadLength = buffer.remaining();
            byte[] encryptedPayload = new byte[payloadLength];
            buffer.get(encryptedPayload);

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

            // 5. Dekripsi data menggunakan 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("Gagal mendekripsi payload Kafka: ", 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]; // Simulasi 256-bit KEK
        secureRandom.nextBytes(localMasterKek);
    }

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

        // 2. Enkripsi (Wrap) DEK menggunakan localMasterKek (KEK)
        byte[] encryptedKey = wrapKey(plaintextKey);

        return new KmsDekEnvelope(plaintextKey, encryptedKey);
    }

    public byte[] decryptDataKey(byte[] encryptedKey) {
        // Dekripsi (Unwrap) DEK menggunakan 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;
    }
}