Безопасность кластеров Apache Kafka

import javax.crypto.Cipher;
import javax.crypto.spec.SecretKeySpec;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
 
public class KafkaMessageEncryptor {
    private final SecretKeySpec secretKey;
    private static final String ALGORITHM = "AES";
    
    public KafkaMessageEncryptor(String secret) {
        byte[] key = secret.getBytes(StandardCharsets.UTF_8);
        this.secretKey = new SecretKeySpec(key, ALGORITHM);
    }
    
    public String encrypt(String strToEncrypt) {
        try {
            Cipher cipher = Cipher.getInstance(ALGORITHM);
            cipher.init(Cipher.ENCRYPT_MODE, secretKey);
            return Base64.getEncoder().encodeToString(
                cipher.doFinal(strToEncrypt.getBytes(StandardCharsets.UTF_8))
            );
        } catch (Exception e) {
            throw new RuntimeException("Ошибка при шифровании: " + e.getMessage(), e);
        }
    }
    
    public String decrypt(String strToDecrypt) {
        try {
            Cipher cipher = Cipher.getInstance(ALGORITHM);
            cipher.init(Cipher.DECRYPT_MODE, secretKey);
            return new String(
                cipher.doFinal(Base64.getDecoder().decode(strToDecrypt)),
                StandardCharsets.UTF_8
            );
        } catch (Exception e) {
            throw new RuntimeException("Ошибка при расшифровке: " + e.getMessage(), e);
        }
    }
}

// В продюсере
KafkaMessageEncryptor encryptor = new KafkaMessageEncryptor("ваш-секретный-ключ");
String originalMessage = "Конфиденциальная информация";
String encryptedMessage = encryptor.encrypt(originalMessage);
producer.send(new ProducerRecord<>("encrypted-topic", encryptedMessage));
 
// В консьюмере
KafkaMessageEncryptor encryptor = new KafkaMessageEncryptor("ваш-секретный-ключ");
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
    String encryptedMessage = record.value();
    String decryptedMessage = encryptor.decrypt(encryptedMessage);
    System.out.println("Расшифрованное сообщение: " + decryptedMessage);
}

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
 
public class SensitiveDataProcessor {
    private final KafkaMessageEncryptor encryptor;
    private final ObjectMapper mapper = new ObjectMapper();
    private final String[] sensitiveFields = {"creditCard", "ssn", "password"};
    
    public SensitiveDataProcessor(String encryptionKey) {
        this.encryptor = new KafkaMessageEncryptor(encryptionKey);
    }
    
    public String processForSending(String jsonString) throws Exception {
        JsonNode rootNode = mapper.readTree(jsonString);
        for (String field : sensitiveFields) {
            encryptSensitiveField((ObjectNode) rootNode, field);
        }
        return mapper.writeValueAsString(rootNode);
    }
    
    public String processAfterReceiving(String jsonString) throws Exception {
        JsonNode rootNode = mapper.readTree(jsonString);
        for (String field : sensitiveFields) {
            decryptSensitiveField((ObjectNode) rootNode, field);
        }
        return mapper.writeValueAsString(rootNode);
    }
    
    private void encryptSensitiveField(ObjectNode node, String fieldName) {
        if (node.has(fieldName)) {
            String value = node.get(fieldName).asText();
            node.put(fieldName, encryptor.encrypt(value));
        }
    }
    
    private void decryptSensitiveField(ObjectNode node, String fieldName) {
        if (node.has(fieldName)) {
            String value = node.get(fieldName).asText();
            node.put(fieldName, encryptor.decrypt(value));
        }
    }
}

public class KeyRotationManager {
    private final Map<String, String> keyVersions = new ConcurrentHashMap<>();
    private final KafkaMessageEncryptor currentEncryptor;
    private String currentKeyVersion;
    
    public KeyRotationManager(String initialKeyVersion, String initialKey) {
        this.currentKeyVersion = initialKeyVersion;
        this.keyVersions.put(initialKeyVersion, initialKey);
        this.currentEncryptor = new KafkaMessageEncryptor(initialKey);
    }
    
    public synchronized void addNewKeyVersion(String version, String key) {
        keyVersions.put(version, key);
    }
    
    public synchronized void switchToNewKeyVersion(String newVersion) {
        if (!keyVersions.containsKey(newVersion)) {
            throw new IllegalArgumentException("Ключ версии " + newVersion + " не найден");
        }
        currentKeyVersion = newVersion;
        // Обновляем текущий шифровальщик
        String newKey = keyVersions.get(newVersion);
        // Здесь нужно реализовать обновление шифровальщика
    }
    
    public String encryptWithCurrentKey(String data) {
        // Добавляем версию ключа в префикс зашифрованных данных
        String encrypted = currentEncryptor.encrypt(data);
        return currentKeyVersion + ":" + encrypted;
    }
    
    public String decrypt(String encryptedData) {
        // Извлекаем версию ключа из префикса
        String[] parts = encryptedData.split(":", 2);
        if (parts.length != 2) {
            throw new IllegalArgumentException("Неверный формат зашифрованных данных");
        }
        
        String keyVersion = parts[0];
        String encryptedPayload = parts[1];
        
        // Используем соответствующий ключ для расшифровки
        String key = keyVersions.get(keyVersion);
        if (key == null) {
            throw new IllegalStateException("Ключ версии " + keyVersion + " не найден");
        }
        
        KafkaMessageEncryptor encryptor = new KafkaMessageEncryptor(key);
        return encryptor.decrypt(encryptedPayload);
    }
}

public class AuditingAuthorizer extends AclAuthorizer {
    private static final Logger LOG = LoggerFactory.getLogger(AuditingAuthorizer.class);
    
    @Override
    public List<AuthorizationResult> authorize(AuthorizableRequestContext requestContext, 
                                              List<Action> actions) {
        // Записываем в лог все запросы с детальной информацией
        KafkaPrincipal principal = requestContext.principal();
        String clientId = requestContext.clientId();
        InetAddress clientAddress = requestContext.clientAddress();
        
        for (Action action : actions) {
            LOG.info("Запрос авторизации: principal={}, clientId={}, ipAddress={}, " +
                    "resource={}, operation={}", 
                    principal, clientId, clientAddress, 
                    action.resourcePattern(), action.operation());
        }
        
        // Делегируем фактическое решение родительскому классу
        return super.authorize(requestContext, actions);
    }
}

public class AuditProducerInterceptor implements ProducerInterceptor<String, String> {
    private static final Logger LOG = LoggerFactory.getLogger(AuditProducerInterceptor.class);
 
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        LOG.info("Отправка сообщения: topic={}, partition={}, key={}", 
                record.topic(), record.partition(), record.key());
        return record;
    }
 
    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        // Ничего не делаем
    }
 
    @Override
    public void close() {}
 
    @Override
    public void configure(Map<String, ?> configs) {}
}

public class AnomalyDetector {
    private final Map<String, UserProfile> userProfiles = new HashMap<>();
    private final double anomalyThreshold = 3.0; // Стандартное отклонение
    
    public boolean isAnomaly(String user, String topic, long messageSize, 
                           LocalDateTime accessTime) {
        UserProfile profile = userProfiles.computeIfAbsent(user, k -> new UserProfile());
        
        // Вычисляем отклонение от нормы
        double sizeDeviation = (messageSize - profile.getAvgMessageSize()) 
                              / profile.getMessageSizeStdDev();
        
        // Проверяем время доступа
        boolean unusualTime = !profile.isUsualAccessTime(accessTime);
        
        // Проверяем доступ к нетипичным топикам
        boolean unusualTopic = !profile.getFrequentTopics().contains(topic);
        
        // Обновляем профиль
        profile.updateStats(topic, messageSize, accessTime);
        
        return sizeDeviation > anomalyThreshold || (unusualTime && unusualTopic);
    }
}