6.3. 인증

프로세스

1. 사용자 인증서에 서명하는 데 사용되는 CA(Certification Authority)의 공개 키가 포함된 JKS(Java 키 저장소) 신뢰 저장소를 준비합니다.

2. 다음과 같이 모든 클러스터 노드에서 Kafka 구성 속성 파일을 편집합니다.

   • ssl.truststore.location 속성을 사용하여 JKS 신뢰 저장소의 경로를 지정합니다.
   • 신뢰 저장소가 암호로 보호되는 경우 ssl.truststore.password 속성을 사용하여 암호를 설정합니다.

   • ssl.client.auth 속성을 required 로 설정합니다.

     TLS 클라이언트 인증 구성

     ssl.truststore.location=/path/to/truststore.jks
     ssl.truststore.password=123456
     ssl.client.auth=required

3. Kafka 브로커를 시작합니다.

프로세스

1. /opt/kafka/config/jaas.conf JAAS 구성 파일을 편집하거나 생성하여 PlainLoginModule 을 활성화하고 허용된 사용자 이름과 암호를 지정합니다.

   이 파일이 모든 Kafka 브로커에서 동일한지 확인합니다.

   JAAS 구성

   KafkaServer {
       org.apache.kafka.common.security.plain.PlainLoginModule required
       user_admin="123456"
       user_user1="123456"
       user_user2="123456";
   };

2. 다음과 같이 모든 클러스터 노드에서 Kafka 구성 속성 파일을 편집합니다.

   • listener.security.protocol.map 속성을 사용하여 특정 리스너에서 SASL PLAIN 인증을 활성화합니다. SASL_PLAINTEXT 또는 SASL_SSL 을 지정합니다.

   • sasl.enabled.mechanisms 속성을 PLAIN 으로 설정합니다.

     SASL 일반 구성

     listeners=INSECURE://:9092,AUTHENTICATED://:9093,REPLICATION://:9094
     listener.security.protocol.map=INSECURE:PLAINTEXT,AUTHENTICATED:SASL_PLAINTEXT,REPLICATION:PLAINTEXT
     sasl.enabled.mechanisms=PLAIN

3. (re) KAFKA_OPTS 환경 변수를 사용하여 Kafka 브로커를 시작하여 JAAS 구성을 Kafka 브로커에 전달합니다.

   su - kafka
   export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties

프로세스

1. /opt/kafka/config/jaas.conf JAAS 구성 파일을 편집하거나 생성하여 ScramLoginModule 을 활성화합니다.

   이 파일이 모든 Kafka 브로커에서 동일한지 확인합니다.

   JAAS 구성

   KafkaServer {
       org.apache.kafka.common.security.scram.ScramLoginModule required;
   };

2. 다음과 같이 모든 클러스터 노드에서 Kafka 구성 속성 파일을 편집합니다.

   • listener.security.protocol.map 속성을 사용하여 특정 리스너에서 SASL SCRAM 인증을 활성화합니다. SASL_PLAINTEXT 또는 SASL_SSL 을 지정합니다.

   • sasl.enabled.mechanisms 옵션을 SCRAM-SHA-256 또는 SCRAM-SHA-512 로 설정합니다.

     예를 들면 다음과 같습니다.

     listeners=INSECURE://:9092,AUTHENTICATED://:9093,REPLICATION://:9094
     listener.security.protocol.map=INSECURE:PLAINTEXT,AUTHENTICATED:SASL_PLAINTEXT,REPLICATION:PLAINTEXT
     sasl.enabled.mechanisms=SCRAM-SHA-512

3. (re) KAFKA_OPTS 환경 변수를 사용하여 Kafka 브로커를 시작하여 JAAS 구성을 Kafka 브로커에 전달합니다.

   su - kafka
   export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties

프로세스

1. sasl.mechanism.inter.broker.protocol 속성을 사용하여 Kafka 구성에서 inter-broker SASL 메커니즘을 지정합니다.

   broker SASL 메커니즘

   sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512

2. username 및 password 필드를 사용하여 KafkaServer JAAS 컨텍스트에서broker 간 통신의 사용자 이름과 암호를 지정합니다.

   broker JAAS 컨텍스트

   KafkaServer {
       org.apache.kafka.common.security.plain.ScramLoginModule required
       username="kafka"
       password="changeit"
       # ...
   };

1. 서비스 주체
2. Kerberos 로그인을 사용하는 Kafka 브로커
3. Kerberos 인증을 사용하여 Kafka에 액세스할 수 있는 생산자 및 소비자 클라이언트

1. Kerberos KDC를 통해 주 키를 저장하는 keytab과 서비스 주체를 생성합니다.

   Kerberos 주체의 도메인 이름이 대문자인지 확인합니다.

   예를 들면 다음과 같습니다.

   • kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM
   • producer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM
   • consumer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM

2. 호스트에 디렉터리를 만들고 keytab 파일을 추가합니다.

   예를 들면 다음과 같습니다.

   /opt/kafka/krb5/kafka-node1.keytab
   /opt/kafka/krb5/kafka-producer1.keytab
   /opt/kafka/krb5/kafka-consumer1.keytab

3. kafka 사용자가 디렉터리에 액세스할 수 있는지 확인합니다.

   chown kafka:kafka -R /opt/kafka/krb5

1. 다음 요소를 사용하여 opt/kafka/config/jaas.conf 파일을 수정합니다.

   KafkaServer {
       com.sun.security.auth.module.Krb5LoginModule required
       useKeyTab=true
       storeKey=true
       keyTab="/opt/kafka/krb5/kafka-node1.keytab"
       principal="kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };
   KafkaClient {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       useTicketCache=false
       keyTab="/opt/kafka/krb5/kafka-node1.keytab"
       principal="kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };

2. 리스너가 SASL/GSSAPI 로그인을 사용하도록 config/server.properties 파일에서 리스너 구성을 수정하여 Kafka 클러스터에서 각 브로커를 구성합니다.

   SASL 프로토콜을 리스너의 보안 프로토콜 맵에 추가하고 원하지 않는 프로토콜을 제거합니다.

   예를 들면 다음과 같습니다.

   # ...
   broker.id=0
   # ...
   listeners=SECURE://:9092,REPLICATION://:9094 1
   inter.broker.listener.name=REPLICATION
   # ...
   listener.security.protocol.map=SECURE:SASL_PLAINTEXT,REPLICATION:SASL_PLAINTEXT 2
   # ..
   sasl.enabled.mechanisms=GSSAPI 3
   sasl.mechanism.inter.broker.protocol=GSSAPI 4
   sasl.kerberos.service.name=kafka 5
   # ...

   1

   두 개의 리스너: 클라이언트와의 범용 통신을 위한 보안 리스너( communication을 위해 TLS 지원) 및 브루커 간 통신을 위한 복제 리스너가 구성됩니다.

   2

   TLS 지원 리스너의 경우 프로토콜 이름은 SASL_PLAINTEXT입니다. TLS 지원 커넥터의 경우 프로토콜 이름은 SASL_PLAINTEXT입니다. SSL이 필요하지 않은 경우 ssl.* 속성을 제거할 수 있습니다.

   3

   Kerberos 인증을 위한 SASL 메커니즘은 GSSAPI 입니다.

   4

   브랜드 간 통신을 위한 Kerberos 인증.

   5

   인증 요청에 사용되는 서비스의 이름은 동일한 Kerberos 구성을 사용하는 다른 서비스와 구별하기 위해 지정됩니다.

3. JVM 매개변수를 사용하여 Kafka 브로커를 시작하여 Kerberos 로그인 구성을 지정합니다.

   su - kafka
   export KAFKA_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties

4. Kerberos 인증을 사용하도록 Kafka 생산자 및 소비자 클라이언트 구성

1. Kerberos 구성을 생산자 또는 소비자 구성 파일에 추가합니다.

   예를 들면 다음과 같습니다.

   /opt/kafka/config/producer.properties

   # ...
   sasl.mechanism=GSSAPI 1
   security.protocol=SASL_PLAINTEXT 2
   sasl.kerberos.service.name=kafka 3
   sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \ 4
       useKeyTab=true  \
       useTicketCache=false \
       storeKey=true  \
       keyTab="/opt/kafka/krb5/producer1.keytab" \
       principal="producer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   # ...

   1

   Kerberos(GSSAPI) 인증을 위한 구성

   2

   Kerberos는 SASL 일반 텍스트(사용자 이름/암호) 보안 프로토콜을 사용합니다.

   3

   Kerberos KDC에 구성된 Kafka의 서비스 주체(사용자)입니다.

   4

   jaas.conf 에 정의된 것과 동일한 속성을 사용하는 JAAS 구성

   /opt/kafka/config/consumer.properties

   # ...
   sasl.mechanism=GSSAPI
   security.protocol=SASL_PLAINTEXT
   sasl.kerberos.service.name=kafka
   sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
       useKeyTab=true  \
       useTicketCache=false \
       storeKey=true  \
       keyTab="/opt/kafka/krb5/consumer1.keytab" \
       principal="consumer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   # ...

2. 클라이언트를 실행하여 Kafka 브로커에서 메시지를 보내고 받을 수 있는지 확인합니다.

   생산자 클라이언트:

   export KAFKA_HEAP_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Dsun.security.krb5.debug=true"; /opt/kafka/bin/kafka-console-producer.sh --producer.config /opt/kafka/config/producer.properties  --topic topic1 --bootstrap-server node1.example.redhat.com:9094

   소비자 클라이언트:

   export KAFKA_HEAP_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Dsun.security.krb5.debug=true"; /opt/kafka/bin/kafka-console-consumer.sh --consumer.config /opt/kafka/config/consumer.properties  --topic topic1 --bootstrap-server node1.example.redhat.com:9094