Kafka-certificate-signature-serde#
Реализует интерфейсы Serializer/Deserializer, позволяет подписывать/проверять подпись сообщения (record.value()) с помощью X509 сертификата.
Предусловия#
Де/сериализаторы могут работать в двух режимах: public-key и certificate-serial.
Последовательность выполнения#
public-key#
Producer-signature-plugin (serializer) получает приватный ключ и сертификат из vault или локального JKS.
Producer-signature-plugin (serializer) сериализует сообщение сериализатором, указанным с помощью настройки
*.serializer.delegate.Producer-signature-plugin (serializer) подписывает value сообщения и добавляет подпись (signature), алгоритм подписи (algorithm) и публичный ключ (public key) сертификата в заголовки сообщения.
Consumer-signature-plugin (deserializer) проверяет подпись сообщения (signature), используя алгоритм (algorithm) и публичный ключ (public key) из заголовков сообщения.
Consumer-signature-plugin (deserializer) десериализует сообщение десериализатором, указанным с помощью настройки
*.deserializer.delegate.
certificate-serial#
Producer-signature-plugin (serializer) получает приватный ключ и сертификат из vault или локального JKS.
Producer-signature-plugin (serializer) сериализует сообщение сериализатором, указанным с помощью настройки
*.serializer.delegate.Producer-signature-plugin (serializer) подписывает value сообщения и добавляет подпись (signature), алгоритм подписи (algorithm) и серийный номер (certificate serial) сертификата в заголовки сообщения.
Consumer-signature-plugin (deserializer) получает публичный ключ сертификата по его серийному номеру, указанному в заголовке сообщения из локального хранилища (JKS или текстового файла) или vault. При запросе сертификата из vault серийный номер дополнительно проверяется по настраиваемому списку разрешенных серийных номеров.
Consumer-signature-plugin (deserializer) проверяет подпись сообщения (signature), используя алгоритм (algorithm) и полученный публичный ключ.
Consumer-signature-plugin (deserializer) десериализует сообщение десериализатором, указанным с помощью настройки
*.deserializer.delegate.
Подключение к Kafka-клиентам#
Добавить актуальную версию плагина в зависимости проекта.
Сконфигурировать kafka-клиенты в соответствии с примерами.
Примеры конфигурации#
Пример конфигурации в режиме public-key#
Пример конфигурации Kafka Producer#
# 1) Подключить сериализатор
value.serializer = ru.sbt.ss.kafka.serialization.CertificateSignatureSerializer
value.serializer.delegate = org.apache.kafka.common.serialization.StringSerializer
# 2) Настроить режим работы сериализатора
value.serializer.signature.certificate.mode = public-key
# 3) Настроить алгоритм подписи
# Алгоритм должен поддерживаться используемым java security provide'ом.
# Алгоритм подписи (в данном случае RSA) должен совпадать с алгоритмом приватного ключа.
value.serializer.signature.certificate.algorithm = SHA256WithRSA
# 4) Настроить хранилище сертификатов в зависимости от типа (jks, pem или vault):
# ДОБАВИТЬ НАСТРОЙКИ ХРАНИЛИЩА СЕРТИФИКАТОВ ИЗ "Примеры конфигурации хранилища сертификатов для kafka-producer"
value.serializer.signature.certificate.keystore.type = change-me
# 5) ОПЦИОНАЛЬНО Настроить имена системных заголовков
# Имя заголовка с подписью сообщения
# value.serializer.signature.attribute = signature
# Префикс для имен системных заголовков
# value.serializer.signature.attributes.prefix = signature.
# Имя загловка с используемым алгоритмом подписи
# value.serializer.signature.certificate.algorithm.attribute=algorithm
# Имя заголовка с публичным ключом для проверки подписи
# value.serializer.signature.certificate.public.key.attribute=public.key
Пример конфигурации Kafka Consumer#
# 1) Подключить десериализатор
value.deserializer = ru.sbt.ss.kafka.serialization.CertificateSignatureDeserializer
value.deserializer.delegate = org.apache.kafka.common.serialization.StringDeserializer
# 2) Настроить режим работы интерсептора
value.deserializer.signature.certificate.mode = public-key
# 3) ОПЦИОНАЛЬНО Настроить удаление системных заголовков из сообщения
# value.deserializer.signature.remove.headers = false
# 4) ОПЦИОНАЛЬНО Настроить режим работы консюмера при ошибках (по умолчанию nullValueWithHeader)
# value.deserializer.signature.mode = nullValueWithHeader
# 5) ОПЦИОНАЛЬНО Настроить имена системных заголовков
# Имя заголовка с подписью сообщения
# value.deserializer.signature.attribute = signature
# Префикс для имен системных заголовков
# value.deserializer.signature.attributes.prefix = signature.
# Имя загловка с используемым алгоритмом подписи
# value.deserializer.signature.certificate.algorithm.attribute=algorithm
# Имя заголовка с публичным ключом для проверки подписи
# value.deserializer.signature.certificate.public.key.attribute=public.key
Пример конфигурации в режиме certificate-serial#
Пример конфигурации Kafka Producer#
# 1) Подключить сериализатор
value.serializer = ru.sbt.ss.kafka.serialization.CertificateSignatureSerializer
value.serializer.delegate = org.apache.kafka.common.serialization.StringSerializer
# 2) Настроить режим работы сериализаторa
value.serializer.signature.certificate.mode = certificate-serial
# 3) Настроить алгоритм подписи
# Алгоритм должен поддерживаться используемым java security provide'ом.
# Алгоритм подписи (в данном случае RSA) должен совпадать с алгоритмом приватного ключа.
value.serializer.signature.certificate.algorithm = SHA256WithRSA
# 4) Настроить хранилище сертификатов в зависимости от типа (jks, pem или vault):
# ДОБАВИТЬ НАСТРОЙКИ ХРАНИЛИЩА СЕРТИФИКАТОВ ИЗ "Примеры конфигурации хранилища сертификатов для kafka-producer"
value.serializer.signature.certificate.keystore.type = change-me
# 5) ОПЦИОНАЛЬНО Настроить имена системных заголовков
# Имя заголовка с подписью сообщения
# value.serializer.signature.attribute = signature
# Префикс для имен системных заголовков
# value.serializer.signature.attributes.prefix = signature.
# Имя загловка с используемым алгоритмом подписи
# value.serializer.signature.certificate.algorithm.attribute=algorithm
# Имя заголовка с серийным номером сертификата для проверки подписи
# value.serializer.signature.certificate.serial.attribute=certificate.serial
Пример конфигурации Kafka Consumer#
# 1) Подключить десериализатор
value.deserializer = ru.sbt.ss.kafka.serialization.CertificateSignatureDeserializer
value.deserializer.delegate = org.apache.kafka.common.serialization.StringDeserializer
# 2) Настроить режим работы десериализатора
value.deserializer.signature.certificate.mode = certificate-serial
# 3) ОПЦИОНАЛЬНО Настроить удаление системных заголовков из сообщения
# value.deserializer.signature.remove.headers = false
# 4) ОПЦИОНАЛЬНО Настроить режим работы консюмера при ошибках (по умолчанию nullValueWithHeader)
# value.deserializer.signature.mode = nullValueWithHeader
# 5) Настроить хранилище сертификатов в зависимости от типа (jks, pem, file или vault):
# ДОБАВИТЬ НАСТРОЙКИ ХРАНИЛИЩА СЕРТИФИКАТОВ ИЗ "Примеры конфигурации хранилища сертификатов для kafka-consumer"
value.deserializer.signature.certificate.truststore.type = change-me
# 6) ОПЦИОНАЛЬНО Настроить имена системных заголовков
# Имя заголовка с подписью сообщения
# value.deserializer.signature.attribute = signature
# Префикс для имен системных заголовков
# value.deserializer.signature.attributes.prefix = signature.
# Имя загловка с используемым алгоритмом подписи
# value.deserializer.signature.certificate.algorithm.attribute=algorithm
# Имя заголовка с серийным номером сертификата для проверки подписи
# value.serializer.signature.certificate.serial.attribute=certificate.serial
Примеры конфигурации хранилища сертификатов для kafka-producer#
Пример конфигурации хранилища типа jks для kafka-producer#
# 4.1) Настроить хранилище сертификатов типа jks
value.serializer.signature.certificate.keystore.type = jks
# По умолчанию используется хранилище сертификатов из настроек kafka client:
ssl.keystore.location = ssl/keystore.jks
ssl.keystore.password = password
ssl.key.password = password
# Сериализатор может использовать отдельное хранилище сертификатов:
# value.serializer.signature.certificate.ssl.keystore.location = ssl/keystore.jks
# value.serializer.signature.certifficate.ssl.keystore.password = password
# value.serializer.signature.certificate.ssl.key.password = password
# Выбрать сертификат из хранилища, указав его alias (по умолчанию используется первый)
# value.serializer.signature.certificate.key.alias =
Пример конфигурации хранилища типа pem для kafka-producer#
# 4.1) Настроить хранилище сертификатов типа pem
# value.serializer.signature.certificate.keystore.type = pem
# value.serializer.signature.certificate.keystore.crt.location = certificate.pem
# value.serializer.signature.certificate.keystore.private.key.location= privateKey.key
Пример конфигурации хранилища типа vault для kafka-producer#
# 4.1) Настроить подключение к vault
value.serializer.signature.certificate.keystore.type = vault
# ВСЕ ПАРАМЕТРЫ НИЖЕ ЯВЛЯЮТСЯ ПАРАМЕТРАМИ ПЛАГИНА ssl-context-builder С ПРЕФИКСОМ `interceptor.signature.certificate.`
# Полный список возможных параметров есть в документации плагина ssl-context-builder
# Адрес vault
value.serializer.signature.certificate.ssl.vault.address = https://host:port
# ОПЦИОНАЛЬНО Namespace vault
# value.serializer.signature.certificate.ssl.vault.namespace = namespace
# ОПЦИОНАЛЬНО Настройки повторной отправки запросов к vault
# Кол-во попыток переотправки запроса
# value.serializer.signature.certificate.ssl.vault.retries = 5
# Тайм-аут отправки запроса, с
# value.serializer.signature.certificate.ssl.vault.timeout = 3
# Интервал между повторными попытками переотправки запроса, мс
# value.serializer.signature.certificate.ssl.vault.retry.interval = 500
# Настройки ssl для подключения к vault
value.serializer.signature.certificate.ssl.vault.tls.enable = true
value.serializer.signature.certificate.ssl.vault.tls.keystore.location = vault-keystore.jks
value.serializer.signature.certificate.ssl.vault.tls.keystore.password = password
value.serializer.signature.certificate.ssl.vault.tls.key.password = password
value.serializer.signature.certificate.ssl.vault.tls.truststore.location = vault-keystore.jks
value.serializer.signature.certificate.ssl.vault.tls.truststore.password = password
value.serializer.signature.certificate.ssl.endpoint.identification.algorithm =
# Настройки авторизации vault
# Пример авторизации с помощью логина и пароля
# Тип авторизации (approle, certificate, password или token)
value.serializer.signature.certificate.ssl.vault.auth.type = password
value.serializer.signature.certificate.ssl.vault.auth.username = test
value.serializer.signature.certificate.ssl.vault.auth.password = password
# Пример авторизации в помощью approle
# value.serializer.signature.certificate.ssl.vault.auth.type = approle
# value.serializer.signature.certificate.ssl.vault.auth.role.id = role
# value.serializer.signature.certificate.ssl.vault.auth.secret.id = secret
# 4.2 Настроить получение сертификата из vault
# Получить сертификат можно двумя способами:
# a) Сгенерировать новый сертификат с помощью pki engine
# b) Получить заранее загруженный сертификат в формате pem с помощью kv engine
# 4.2.a) Настроить генерацию сертификата с помощью vault (pki engine)
# value.serializer.signature.certificate.ssl.vault.pki.mode = pki
#
# ОПЦИОНАЛЬНО Метод получения сертификата:
# * issue (по умолчанию) - выпускает новый сертификат при каждом обращении
# * fetch (движок SberCA) - выпускает новый сертификат только если сертификат с указанными параметрами не существует или истек
# При использовании движка SberCA параметры `ssl.vault.pki.ttl`, `ssl.vault.pki.csr`, `ssl.vault.pki.csr.path` и `ssl.vault.pki.not.after` не поддерживаются и будут проигнорированы.
# value.serializer.signature.certificate.ssl.vault.pki.method = issue
#
# ОПЦИОНАЛЬНО Путь до pki engine
# value.serializer.signature.certificate.ssl.vault.pki.mount = pki
#
# Имя роли для выпуска сертификата
value.serializer.signature.certificate.ssl.vault.pki.role.name = role
#
# Common name сертификата (CN)
value.serializer.signature.certificate.ssl.vault.pki.common.name = INTERCEPTOR-TEST
#
# ОПЦИОНАЛЬНО Электронный адрес владельца сертификата, задается при использовании Secret Manager
# value.serializer.signature.certificate.ssl.vault.pki.email = email@example.com
#
# ОПЦИОНАЛЬНО Alternative names сертификата
# value.serializer.signature.certificate.ssl.vault.pki.alt.names = alt-name
#
# ОПЦИОНАЛЬНО Alternative ip сертификата
# value.serializer.signature.certificate.ssl.vault.pki.alt.ip = 127.0.0.1
#
# ОПЦИОНАЛЬНО TTL (time-to-live) сертификата
# value.serializer.signature.certificate.ssl.vault.pki.ttl =
#
# ОПЦИОНАЛЬНО Запрос на создание сертификата csr (Certificate signing request)
# value.serializer.signature.certificate.ssl.vault.pki.csr =
#
# ОПЦИОНАЛЬНО Путь до файла с запросом на создание сертификата csr (Certificate signing request)
# value.serializer.signature.certificate.ssl.vault.pki.csr.path =
# 4.2.b) Настроить получение заранее загруженного сертификата с помощью kv engine
# value.serializer.signature.certificate.ssl.vault.pki.mode = kv
#
# Указать путь до key-value хранилища с приватным ключем и сертификатом в формате pem
# value.serializer.signature.certificate.ssl.vault.kv.pem.path = kv1/certificate
#
# Указать имя секрета в key-value хранилище, который содержит сертификат в формате pem
# value.serializer.signature.certificate.ssl.vault.kv.pem.name = cert
#
# Указать имя секрета в key-value хранилище, который содержит приватный ключ в формате pem
# value.serializer.signature.certificate.ssl.vault.kv.pem.key = key
# 4.3) ОПЦИОНАЛЬНО Настроить хранилище сертификатов (локальный кэш)
# Пути до хранилищ сертификатов, сгенерированных vault
# Если не указаны - сертификаты будут генерироваться каждый раз при старте приложения
# value.serializer.signature.certificate.ssl.keystore.location = producer-vault-keystore.jks
# value.serializer.signature.certificate.ssl.truststore.location = producer-vault-truststore.jks
# alias клиентского сертификата в keystore
# value.serializer.signature.certificate.ssl.vault.alias.key = key
# alias ca сертификата в truststore
# value.serializer.signature.certificate.ssl.vault.alias.ca = ca
# 4.4) ОПЦИОНАЛЬНО Настроить получение паролей для хранилища сертификатов из vault
# Путь до хранилища секретов в vault
# value.serializer.signature.certificate.ssl.vault.secret.path = kv1/interceptor
# Версия secret engine
# value.serializer.signature.certificate.ssl.vault.engine.version = 1
# Имя секрета, содержащего пароль для private key
# value.serializer.signature.certificate.ssl.vault.secret.key = key
# Имя секрета, содержащего пароль для keystore
# value.serializer.signature.certificate.ssl.vault.secret.keystore = keystore
# Имя секрета, содержащего пароль для truststore
# value.serializer.signature.certificate.ssl.vault.secret.truststore = truststore
# Указать aliases для trust сертификата и путь до trust'a (для jks или pem). Например:
value.serializer.signature.certificate.ssl.vault.pem.trust.path = A/DEV/BPM/CORE/BPMX/KV/IFT-DAPP/BPMX/kafka-signature
value.serializer.signature.certificate.ssl.vault.pem.trust.aliases = custom_crt
Примеры конфигурации хранилища сертификатов для kafka-consumer#
Пример конфигурации хранилища типа jks для kafka-consumer#
value.deserializer.signature.certificate.truststore.type = jks
# По умолчанию используется хранилище сертификатов из настроек kafka client:
ssl.truststore.location = ssl/truststore.jks
ssl.truststore.password = password
# Десериализатор может использовать отдельное хранилище сертификатов:
value.deserializer.signature.certificate.ssl.truststore.location = truststore.jks
value.deserializer.signature.certificate.ssl.truststore.password = password
Пример конфигурации хранилища типа pem для kafka-consumer#
value.deserializer.signature.certificate.truststore.type = pem
# Указать сертификат или директорию с сертификатами
value.deserializer.signature.truststore.crt.location = certificate.pem
# ОПЦИОНАЛЬНО Указать расширение сертификатов для фильтрации файлов в директории. По умолчанию ".pem".
# value.deserializer.signature.truststore.crt.location.filter = .cer
Пример конфигурации хранилища типа file для kafka-consumer#
value.deserializer.signature.certificate.truststore.type = file
# Файл с публичными ключами в формате "серийный номер сертификата" = "публичный ключ в base64"
value.deserializer.signature.certificate.truststore.file = truststore.txt
Пример truststore.txt:
"<hash>" = "MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAiCa/z3VMo7LxIP/e2E6CjIFy5lwTgejw2r3/ykfbPMtM6jF50aU5a0hki2t+PfkvrHpUvy/o5XGTkLmZKBdD2JpX4FIvf/ZQtC/yvakTsP5//Tgkb7QaEa66xSk03dSHVGmZ/nr+d6x7kfa2vzfZd8gAgijwssZnACWT74NjlF3slsEZdD/S6C1nqm7EDBjd0YC67KXUVGreEnMq0+q5mj5TlQsya7Dglvouz6isn7O/1SsBihtswJ0bKjpk9X3nJUSX2/eHPUlLM23Mmtj9gQOoNO/u90f5b5Vp5dJ1PUPd8xRFBep1Bu1S6HZsqqe80KAf8FyPmG5RgADyPXBXCQIDAQAB"
"<hash>" = "MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAn/cOADn/ScshW5F0m3cAlwKY2DxELZ274BDNHjeFVM04w2KEqA0djmP4PWpj742EbO5ZnpfDAu/67Qt8Xia5sfUpDes6hqPnETrqq96JeLnhtKXiTN9cmfBbKlx7neZI0elfl450tCNmfUOlnbAVvNLPSvj8/1Dem1PxHIBCtHkjK3P0D/76Bewz7oLkPBH47XUBBpwpNEYOx8sFaze6C123a/l99RJJLhiZ+GaVo6BKxcQCS3Z40gG+rHdxALvnMEj0MP0K2/8Q8W2IPkg4JReny3wPkuQ53cmNMMfhl7jr0WBzxXICesLjeaTqrfaFlT9oxcTcwK9PCTlAhWvtEQIDAQAB"
Пример конфигурации хранилища типа vault для kafka-consumer#
# 4.1) Настроить подключение к vault
value.deserializer.signature.certificate.keystore.type = vault
# ВСЕ ПАРАМЕТРЫ НИЖЕ ЯВЛЯЮТСЯ ПАРАМЕТРАМИ ПЛАГИНА ssl-context-builder С ПРЕФИКСОМ `interceptor.signature.certificate.`
# Полный список возможных параметров есть в документации плагина ssl-context-builder
# Адрес vault
value.deserializer.signature.certificate.ssl.vault.address = https://host:port
# ОПЦИОНАЛЬНО Namespace vault
# value.deserializer.signature.certificate.ssl.vault.namespace = namespace
# ОПЦИОНАЛЬНО Настройки повторной отправки запросов к vault
# Кол-во попыток переотправки запроса
# value.deserializer.signature.certificate.ssl.vault.retries = 5
# Тайм-аут отправки запроса, с
# value.deserializer.signature.certificate.ssl.vault.timeout = 3
# Интервал между повторными попытками переотправки запроса, мс
# value.deserializer.signature.certificate.ssl.vault.retry.interval = 500
# Настройки ssl для подключения к vault
value.deserializer.signature.certificate.ssl.vault.tls.enable = true
value.deserializer.signature.certificate.ssl.vault.tls.keystore.location = vault-keystore.jks
value.deserializer.signature.certificate.ssl.vault.tls.keystore.password = password
value.deserializer.signature.certificate.ssl.vault.tls.key.password = password
value.deserializer.signature.certificate.ssl.vault.tls.truststore.location = vault-keystore.jks
value.deserializer.signature.certificate.ssl.vault.tls.truststore.password = password
value.deserializer.signature.certificate.ssl.endpoint.identification.algorithm =
# Настройки авторизации vault
# Пример авторизации с помощью логина и пароля
# Тип авторизации (approle, certificate, password или token)
value.deserializer.signature.certificate.ssl.vault.auth.type = password
value.deserializer.signature.certificate.ssl.vault.auth.username = test
value.deserializer.signature.certificate.ssl.vault.auth.password = password
# Пример авторизации в помощью approle
# value.deserializer.signature.certificate.ssl.vault.auth.type = approle
# value.deserializer.signature.certificate.ssl.vault.auth.role.id = role
# value.deserializer.signature.certificate.ssl.vault.auth.secret.id = secret
# 4.2) ОПЦИОНАЛЬНО Настроить хранилище сертификатов (локальный кэш)
# Путь до хранилища сертификатов, поулченных из vault
# value.deserializer.signature.certificate.ssl.truststore.location = consumer-vault-truststore.jks
# 4.3) ОПЦИОНАЛЬНО Настроить получение паролей для хранилища сертификатов из vault
# Путь до хранилища секретов в vault
# value.deserializer.signature.certificate.ssl.vault.secret.path = kv1/interceptor
# Версия secret engine
# value.deserializer.signature.certificate.ssl.vault.engine.version = 1
# Имя секрета, содержащего пароль для truststore
# value.deserializer.signature.certificate.ssl.vault.secret.truststore = truststore
# 4.4) ОПЦИОНАЛЬНО Настроить список доверенных сертификатов
# По умолчанию все сертификаты являются доверенными
# Если сертификат не входит в этот список - будет получена ошибка проверки подписи.
#
# Можно использовать один из трех вариантов проверки сертификата:
# 4.4.1) Проверка по серийному номеру сертификата
# Атрибут сертификата, который будет проверяться по списку доверенных сертификатов (по умолчанию serial):
# value.deserializer.signature.certificate.allowed.certificate.list.type = serial
#
# Список серийных номеров строкой через запятую:
# value.deserializer.signature.certificate.allowed.certificate.list = 3ca7322c, serial
# Путь до файла, в файле каждая строка является отдельным серийным номером:
# value.deserializer.signature.certificate.allowed.certificate.list.file = allowed-serials.txt
#
# DEPRECATED Также для типа serial существуют устаревшие настройки, аналогичные настройкам выше:
# Настройки allowed.certificate.list.* имеют наивысший приоритет
# value.deserializer.signature.certificate.allowed.serial = 3ca7322c, serial
# value.deserializer.signature.certificate.allowed.serial.file = allowed-serials.txt
#
# 4.4.2) Проверка по Common Name (CN) сертификата
# Атрибут сертификата, который будет проверяться по списку доверенных сертификатов:
# value.deserializer.signature.certificate.allowed.certificate.list.type = cn
#
# Список CN сертификатов строкой через запятую
# value.deserializer.signature.certificate.allowed.certificate.list = CN=test, CN=test2
# Путь до файла, в файле каждая строка является отдельным CN сертификата:
# value.deserializer.signature.certificate.allowed.certificate.list.file = allowed-cn.txt
#
# 4.4.3) Проверка по Distinguished Name (DN) сертификата
# Атрибут сертификата, который будет проверяться по списку доверенных сертификатов:
# value.deserializer.signature.certificate.allowed.certificate.list.type = dn
#
# Список DN сертификатов строкой через точку с запятой
# value.deserializer.signature.certificate.allowed.certificate.list = CN=test, OU=FPSS, O=SBT, ST=Moscow, C=RU; CN=test2, C=RU
# Путь до файла, в файле каждая строка является отдельным CN сертификата:
# value.deserializer.signature.certificate.allowed.certificate.list.file = allowed-dn.txt
Поведение при ошибках#
В случае, если сообщение не прошло проверку подписи при получении (consumer.poll()) поведение настраивается параметром value.deserializer.mode:
nullValueWithHeader(используется по умолчанию) – клиент получит сообщение, которое содержит null вместо value и дополнительный заголовокvalidation.errorc сообщением об ошибке;failOnConsume– методconsumer.poll()выбросит исключение, клиент не получит ни одного сообщения из пачки.
Загрузка хранилища сертификатов из classpath#
Для загрузки хранилища сертификатов из classpath необходимо указать протокол classpath:// в пути до файла, например:
value.serializer.signature.certificate.ssl.keystore.location = classpath://ssl/keystore.jks
Не работает для стандартных ssl-настроек kafka-client типа ssl.keystore.location
Не работает при использовании pem сертификатов
Использование Conscrypt#
Conscrypt – open-source java security provider, его реализация алгоритмов подписи RSA в тестах показывает двухкратный прирост производительности по сравнению со стандартным провайдером JVM.
Подключение к kafka-producer#
Библиотека Conscrypt уже включена в транзитивные зависимости.
Есть два способа подключить conscrypt:
С помощью параметра
conscrypt.enabled(по умолчанию данный параметр уже активирован):
value.serializer.signature.certificate.conscrypt.enabled = true
(Устаревший способ) Инициализировать провайдер перед запуском kafka-producer:
import org.conscrypt.Conscrypt;
...
Security.insertProviderAt(Conscrypt.newProvider(), Security.getProviders.length);
2.1 Указать провайдер в конфигурации продюсера:
# Имя java security provider'а, используемого для подписи сообщений
value.serializer.signature.certificate.ssl.provider = Conscrypt
Использование Conscrypt для проверки подписи на консьюмер не дает значительного прироста производительности.
Результат#
Выполнено подключение перехватчика Kafka-certificate-signature-serde.