add kafka mtls info (#38246)

Co-authored-by: Irina Skvortsova <[email protected]>
Co-authored-by: sintjuri <[email protected]>
Co-authored-by: Andrey Fomichev <[email protected]>

5 files changed, 325 insertions, 8 deletions

diff --git a/ydb/docs/en/core/reference/kafka-api/auth.md b/ydb/docs/en/core/reference/kafka-api/auth.md
index d837d4456ee..2330c21146b 100644
--- a/ydb/docs/en/core/reference/kafka-api/auth.md
+++ b/ydb/docs/en/core/reference/kafka-api/auth.md
@@ -10,7 +10,9 @@ Authentication is always enabled when using the [Kafka API in Yandex Cloud](http
 
 ## How does authentication work in the Kafka API?
 
-The Kafka API uses the `SASL_PLAINTEXT/PLAIN` or `SASL_SSL/PLAIN` authentication mechanism.
+The Kafka API uses the `SASL_PLAINTEXT/PLAIN`, `SASL_SSL/PLAIN`and mTLS authentication mechanisms.
+
+### Auhentication using PLAIN or SCRAM-SHA-256
 
 The following variables are required for authentication:
 
@@ -29,4 +31,159 @@ The `<sasl.username>` and `<sasl.password>` parameters are formed differently. S
 
 {% endnote %}
 
-For authentication examples, see [Kafka API usage examples](./examples.md).
\ No newline at end of file
+For authentication examples, see [Kafka API usage examples](./examples.md).
+
+### Authentication using mTLS
+To enable mTLS authentication, the following steps are required.
+
+#### Server and client certificates creation
+
+For each step below the examples of commands are given. Substitute \*\*\* with your values.
+
+1. Create Certificate Authority (CA)
+
+```bash
+openssl genrsa -out ca-key.pem 4096
+```
+
+```bash
+openssl req -new -x509 -days 3650 -key ca-key.pem -out ca-cert.pem -subj "/C=***/ST=***/L=***/O=***/CN=MyKafkaRootCA"
+```
+
+2. Create server certificate
+
+```bash
+openssl genrsa -out server-key.pem 4096
+```
+
+In the next command put your host name instead of `serverhost.com`.
+
+```bash
+openssl req -new -key server-key.pem -out server-cert.csr -subj "/C=***/ST=***/L=***/O=***/CN=serverhost.com"
+```
+
+```bash
+cat > server-ext.cnf << EOF
+authorityKeyIdentifier=keyid,issuer
+basicConstraints=CA:FALSE
+keyUsage = digitalSignature, nonRepudiation, keyEncipherment, dataEncipherment
+subjectAltName = DNS:serverhost.com
+EOF
+```
+
+```bash
+openssl x509 -req -in server-cert.csr -CA ca-cert.pem -CAkey ca-key.pem -CAcreateserial -out server-cert.pem -days 365 -extfile server-ext.cnf
+```
+
+3. Create client certificate
+
+```bash
+openssl genrsa -out client-key.pem 4096
+```
+
+Substitute `clienthost.com` with hostname of your client.
+
+```bash
+openssl req -new -key client-key.pem -out client-cert.csr -subj "/C=***/ST=***/L=***/O=***/CN=clienthost.com"
+```
+
+```bash
+cat > client-ext.cnf << EOF
+authorityKeyIdentifier=keyid,issuer
+basicConstraints=CA:FALSE
+keyUsage = digitalSignature, nonRepudiation, keyEncipherment, dataEncipherment
+subjectAltName = DNS:clienthost.com
+EOF
+```
+
+```bash
+openssl x509 -req -in client-cert.csr -CA ca-cert.pem -CAkey ca-key.pem -CAcreateserial -out client-cert.pem -days 365 -extfile client-ext.cnf
+```
+
+
+4. Add created certificates to keystore and truststore
+
+For server:
+
+```bash
+openssl pkcs12 -export -in server-cert.pem -inkey server-key.pem -out server.p12 -name kafka-server -CAfile ca-cert.pem -caname root -password pass:changeit
+
+keytool -importkeystore -deststorepass changeit -destkeystore server.keystore.jks -srckeystore server.p12 -srcstoretype PKCS12 -srcstorepass changeit -alias kafka-server 
+
+keytool -import -trustcacerts -alias ca -file ca-cert.pem -keystore server.truststore.jks -storepass changeit -noprompt
+```
+
+For client:
+
+```bash
+openssl pkcs12 -export -in client-cert.pem -inkey client-key.pem -out client.p12 -name kafka-client -CAfile ca-cert.pem -caname root -password pass:changeit
+
+keytool -importkeystore -deststorepass changeit -destkeystore client.keystore.jks -srckeystore client.p12 -srcstoretype PKCS12 -srcstorepass changeit -alias kafka-client  
+
+keytool -import -trustcacerts -alias ca -file ca-cert.pem -keystore client.truststore.jks -storepass changeit -noprompt  
+```
+
+After fulfilling these steps you should obtain keystore and truststore, as well as files with certificates and keys.
+
+#### Client configuration
+##### Java SDK example
+```java
+props.put("security.protocol", "SSL");
+props.put("ssl.truststore.password", "changeit");
+props.put("ssl.truststore.location", "/full/path/to/client.truststore.jks");
+props.put("ssl.keystore.location", "/full/path/to/client.keystore.jks");
+props.put("ssl.keystore.password", "changeit");
+props.put("ssl.key.password", "changeit");
+props.put("ssl.endpoint.identification.algorithm", "");
+```
+
+##### Kafka cli example
+```
+security.protocol=SSL
+ssl.truststore.password=changeit
+ssl.truststore.location=/full/path/to/client.truststore.jks
+ssl.keystore.location=/full/path/to/client.keystore.jks
+ssl.keystore.password=changeit
+ssl.key.password=changeit
+ssl.endpoint.identification.algorithm=
+```
+
+#### YDB configuration
+
+It is necessary to specify the required fields in the [kafka_proxy_config](../configuration/kafka.md).
+
+```yaml
+kafka_proxy_config:
+  enable_kafka_proxy: true
+  listening_port: your_port
+
+  mtls_enable: true
+  key: "server-key.pem" # укажите правильные пути до файлов
+  cert: "server-cert.pem"
+  ca: "ca-cert.pem"
+  enable_self_signed_certs: true # разрешаете ли вы самоподписанные сертификаты
+```
+
+In the [client_certificate_authorization](../configuration/client_certificate_authorization.md), specify the following authentication rules:
+
+```yaml
+client_certificate_authorization:
+  client_certificate_definitions:
+    - require_same_issuer: true
+      subject_terms:
+        - short_name: CN
+          suffixes:
+            - '.myhost.net' # нужно заменить на нужный суффикс
+      member_groups:
+        - user@cert # заменить на нужную member группу
+  request_client_certificate: true
+```
+
+Additionally, for proper operation, you need to use the same certificate as in the gRPC settings, so you need to specify the path to this server certificate in the grpc configuration.
+
+Currently, it is not possible to configure Kafka and gRPC with different server certificates or to specify a server certificate only in the kafka_proxy_config settings when using mTLS.
+
+```yaml
+grpc_config:
+  cert: "/path/to/server-cert.pem"
+```
diff --git a/ydb/docs/en/core/reference/kafka-api/constraints.md b/ydb/docs/en/core/reference/kafka-api/constraints.md
index e5fb1590f30..563d4243abd 100644
--- a/ydb/docs/en/core/reference/kafka-api/constraints.md
+++ b/ydb/docs/en/core/reference/kafka-api/constraints.md
@@ -2,9 +2,8 @@
 

 {{ ydb-short-name }} supports [Apache Kafka protocol](https://kafka.apache.org/protocol.html) version 3.4.0 with the following constraints:

 

-1. Only [SASL/PLAIN authentication](https://kafka.apache.org/documentation/#security_sasl) is supported.

 1. [Message compression](https://www.confluent.io/blog/apache-kafka-message-compression) is not supported.

 1. [The topic deletion operation](https://kafka.apache.org/protocol#The_Messages_DeleteTopics) is not supported. To delete a topic, use [YQL](../../yql/reference/syntax/drop-topic.md) or [{{ ydb-short-name }} CLI](../ydb-cli/topic-drop.md).

 1. [CRC checks](https://kafka.apache.org/documentation/#consumerconfigs_check.crcs) are not supported.

 1. [Support for ACL](https://kafka.apache.org/documentation/#security_authz) is not provided. Use [YQL](../../yql/reference/syntax/grant.md) to manage access to topics.

-1. If [auto-partitioning](../../concepts/datamodel/topic.md#autopartitioning) is enabled on a topic, you cannot write to or read from such a topic using the Kafka API.
\ No newline at end of file
+1. If [auto-partitioning](../../concepts/datamodel/topic.md#autopartitioning) is enabled on a topic, you cannot write to or read from such a topic using the Kafka API.

diff --git a/ydb/docs/ru/core/reference/configuration/kafka_proxy_config.md b/ydb/docs/ru/core/reference/configuration/kafka_proxy_config.md
index 8ede0306329..38823a02c63 100644
--- a/ydb/docs/ru/core/reference/configuration/kafka_proxy_config.md
+++ b/ydb/docs/ru/core/reference/configuration/kafka_proxy_config.md
@@ -14,7 +14,10 @@
 || `topic_creation_default_partitions` | uint32 | `1` | Количество партиций, которое будет создано, если при добавлении топика по Kafka протоколу не было указано количество партиций. Аналог опиции [num.partitions](https://kafka.apache.org/documentation/#brokerconfigs_num.partitions) в Apache Kafka. ||
 || `ssl_cerificate` | string | - | Путь к файлу сертификата для доступа по SSL, включающий в себя сразу и файл сертификата и файл ключа. При указании этого параметра Kafka Proxy автоматически начинает обрабатывать запросы с использованием указанного SSL-сертификата. ||
 || `cert` | string | - | Путь к файлу сертификата для доступа по SSL. При указании этого параметра Kafka Proxy автоматически начинает обрабатывать запросы с использованием указанного SSL-сертификата. ||
-|| `key` | string | - | Путь к файлу ключа для доступа по SSL. |#
+|| `key` | string | - | Путь к файлу ключа для доступа по SSL. ||
+|| `ca` | string | - | Путь к файлу certificate authority для mTLS. ||
+|| `enable_self_signed_certs` | bool | `false` | Разрешает использование самоподписанных сертификатов для mTLS. ||
+|| `mtls_enable` | bool | `false` | Включает использование mTLS аутентификации. ||
 
 ## Пример заполненного конфига
 
diff --git a/ydb/docs/ru/core/reference/kafka-api/auth.md b/ydb/docs/ru/core/reference/kafka-api/auth.md
index 12413149992..5e32fae4230 100644
--- a/ydb/docs/ru/core/reference/kafka-api/auth.md
+++ b/ydb/docs/ru/core/reference/kafka-api/auth.md
@@ -9,9 +9,12 @@
 
 Аутентификация всегда включена при использовании [Kafka API в Yandex Cloud](https://yandex.cloud/ru/docs/data-streams/kafkaapi/auth)
 
-## Механизм аутентификации
+## Механизмы аутентификации
+
+В Kafka API поддержано два механизма SASL аутентификации: `PLAIN` и `SCRAM-SHA-256`, а также mTLS аутентификация.
+
+### Аутентификация с помощью PLAIN и SCRAM-SHA-256
 
-В Kafka API поддержано два механизма SASL аутентификации: `PLAIN` и `SCRAM-SHA-256`.
 Оба механизма могут осуществляться как внутри протокола `TLS`, так и вне, порождая соответственно комбинации:
 
 * `SASL_PLAINTEXT/PLAIN`;
@@ -47,3 +50,159 @@
 {% endnote %}
 
 Примеры аутентификации смотрите в [Чтение и запись](./examples.md).
+
+### Аутентификация по mTLS
+
+Чтобы Kafka-клиент мог аутентифицироваться по mTLS, нужно выполнить следующие шаги.
+
+#### Создание серверного и клиентского сертификатов
+
+Для каждого шага ниже представлены примеры команд. Замените \*\*\* на ваши значения.
+
+1. Создайте Certificate Authority (CA)
+
+```bash
+openssl genrsa -out ca-key.pem 4096
+```
+
+```bash
+openssl req -new -x509 -days 3650 -key ca-key.pem -out ca-cert.pem -subj "/C=***/ST=***/L=***/O=***/CN=MyKafkaRootCA"
+```
+
+2. Создайте сертификат для сервера
+
+```bash
+openssl genrsa -out server-key.pem 4096
+```
+
+В следующей команде вместо `serverhost.com` также укажите название вашего хоста.
+
+```bash
+openssl req -new -key server-key.pem -out server-cert.csr -subj "/C=***/ST=***/L=***/O=***/CN=serverhost.com"
+```
+
+```bash
+cat > server-ext.cnf << EOF
+authorityKeyIdentifier=keyid,issuer
+basicConstraints=CA:FALSE
+keyUsage = digitalSignature, nonRepudiation, keyEncipherment, dataEncipherment
+subjectAltName = DNS:serverhost.com
+EOF
+```
+
+```bash
+openssl x509 -req -in server-cert.csr -CA ca-cert.pem -CAkey ca-key.pem -CAcreateserial -out server-cert.pem -days 365 -extfile server-ext.cnf
+```
+
+3. Создайте клиентский сертификат
+
+```bash
+openssl genrsa -out client-key.pem 4096
+```
+
+Замените `clienthost.com` на hostname вашего клиента.
+
+```bash
+openssl req -new -key client-key.pem -out client-cert.csr -subj "/C=***/ST=***/L=***/O=***/CN=clienthost.com"
+```
+
+```bash
+cat > client-ext.cnf << EOF
+authorityKeyIdentifier=keyid,issuer
+basicConstraints=CA:FALSE
+keyUsage = digitalSignature, nonRepudiation, keyEncipherment, dataEncipherment
+subjectAltName = DNS:clienthost.com
+EOF
+```
+
+```bash
+openssl x509 -req -in client-cert.csr -CA ca-cert.pem -CAkey ca-key.pem -CAcreateserial -out client-cert.pem -days 365 -extfile client-ext.cnf
+```
+
+
+4. Добавьте сертификаты в keystore и truststore
+
+Для сервера:
+
+```bash
+openssl pkcs12 -export -in server-cert.pem -inkey server-key.pem -out server.p12 -name kafka-server -CAfile ca-cert.pem -caname root -password pass:changeit
+
+keytool -importkeystore -deststorepass changeit -destkeystore server.keystore.jks -srckeystore server.p12 -srcstoretype PKCS12 -srcstorepass changeit -alias kafka-server 
+
+keytool -import -trustcacerts -alias ca -file ca-cert.pem -keystore server.truststore.jks -storepass changeit -noprompt
+```
+
+Для клиента:
+
+```bash
+openssl pkcs12 -export -in client-cert.pem -inkey client-key.pem -out client.p12 -name kafka-client -CAfile ca-cert.pem -caname root -password pass:changeit
+
+keytool -importkeystore -deststorepass changeit -destkeystore client.keystore.jks -srckeystore client.p12 -srcstoretype PKCS12 -srcstorepass changeit -alias kafka-client  
+
+keytool -import -trustcacerts -alias ca -file ca-cert.pem -keystore client.truststore.jks -storepass changeit -noprompt  
+```
+
+После этих пунктов у вас должны появиться нужные keystore и truststore, а также файлы с сертификатами и ключами.
+
+#### Конфигурация клиента
+##### Пример для Java SDK
+```java
+props.put("security.protocol", "SSL");
+props.put("ssl.truststore.password", "changeit");
+props.put("ssl.truststore.location", "/full/path/to/client.truststore.jks");
+props.put("ssl.keystore.location", "/full/path/to/client.keystore.jks");
+props.put("ssl.keystore.password", "changeit");
+props.put("ssl.key.password", "changeit");
+props.put("ssl.endpoint.identification.algorithm", "");
+```
+
+##### Пример для Kafka cli
+
+```
+security.protocol=SSL
+ssl.truststore.password=changeit
+ssl.truststore.location=/full/path/to/client.truststore.jks
+ssl.keystore.location=/full/path/to/client.keystore.jks
+ssl.keystore.password=changeit
+ssl.key.password=changeit
+ssl.endpoint.identification.algorithm=
+```
+
+#### Конфигурация YDB
+
+Требуется указать нужные поля в конфигурации [kafka_proxy_config](../configuration/kafka_proxy_config.md).
+
+```yaml
+kafka_proxy_config:
+  enable_kafka_proxy: true
+  listening_port: your_port
+
+  mtls_enable: true
+  key: "server-key.pem" # укажите правильные пути до файлов
+  cert: "server-cert.pem"
+  ca: "ca-cert.pem"
+  enable_self_signed_certs: true # разрешаете ли вы самоподписанные сертификаты
+```
+
+Также укажите в конфигурации [client_certificate_authorization](../configuration/client_certificate_authorization.md) правила, по которым будет проходить аутентификация:
+
+```yaml
+client_certificate_authorization:
+  client_certificate_definitions:
+    - require_same_issuer: true
+      subject_terms:
+        - short_name: CN
+          suffixes:
+            - '.myhost.net' # нужно заменить на нужный суффикс
+      member_groups:
+        - user@cert # заменить на нужную member группу
+  request_client_certificate: true
+```
+
+Для корректной работы нужно использовать тот же сертификат, что и в настройках grpc, поэтому нужно указать путь до этого же серверного сертификата в конфигурации grpc.
+
+Сейчас настроить конфигурации Kafka и grpc с разными серверными сертификатами или указать серверный сертификат только в настройках kafka_proxy_config при использовании mtls нельзя.
+```yaml
+grpc_config:
+  cert: "/path/to/server-cert.pem"
+```
diff --git a/ydb/docs/ru/core/reference/kafka-api/constraints.md b/ydb/docs/ru/core/reference/kafka-api/constraints.md
index 08aa3208c5a..7fd7dc970ec 100644
--- a/ydb/docs/ru/core/reference/kafka-api/constraints.md
+++ b/ydb/docs/ru/core/reference/kafka-api/constraints.md
@@ -2,7 +2,6 @@
 
 Поддержка протокола Kafka версии 3.4.0 осуществляется в ограниченном объеме:
 
-1. Поддержаны только SASL/PLAIN и SASL/SCRAM-SHA-256 [механизмы аутентификации](https://kafka.apache.org/documentation/#security_sasl).
 1. Не поддержано [сжатие сообщений](https://www.confluent.io/blog/apache-kafka-message-compression/).
 1. Не поддержана [операция удаления топика](https://kafka.apache.org/protocol#The_Messages_DeleteTopics). Для удаления топика используйте [YQL](../../yql/reference/syntax/drop-topic.md) или [{{ ydb-short-name }} CLI](../ydb-cli/topic-drop.md).
 1. Не поддержана [проверка crc](https://kafka.apache.org/documentation/#consumerconfigs_check.crcs).