kafka와 보안 - SSL

kafka의 통신에 적용해야할 보안 요소는 다음과 같다.

 

SSL 통신

※ kafka는 JAVA로 개발된 플랫폼이어서 jks 로 인증서를 관리하게 만들어주면 kafka가 쉽게 사용할수 있습니다. 이에 따라 jks 로 인증서를 관리하는 체계로 아래에 서술되어 있으니 참고바랍니다.

 

※ 그리고 해당 내용은 kafka server 01번을 CA로 쓰는 프로세스이므로 정석적인 방식은 아닙니다. 사실 인증서 체계는 당연히 분리하는것이 맞습니다. 꼭 kafka 서버 중 하나가 CA가 되어야 하는게 아니며 사용가능한 자원이 한정되어 이렇게 진행하는 것이니 참고바랍니다. 

참고용-jks.pdf

0.42MB

 

 

1. 브로커 키스토어 생성

키스토어는 서버 측면에서 프라이빗 키와 인증서를 저장하며, 자격 증명을 제공합니다. 그외에 프라이빗하고 민감한 정보를 저장합니다. 트로스트 스토어는 일반적으로클라이언트 측면에서 서버가 제공하는 인증서를 검증하기 위한 퍼블릭 키와 서버와 SSL연결에서 유효성을 검사하는 서명된 인증서를 저장하며, 민감한 정보는 저장하지 않습니다. 

# mkdir -p $KAFKA_HOME/ssl
# cd $KAFKA_HOME/ssl
# export SSLPASS=peterpass // 환경변수를 이용해 비밀번호를 설정 후 다음 프로세스 진행

// 키 스토어 생성
# keytool -keystore kafka.server.keystore.jks -alias localhost -keyalg RSA -validity 365 -genkey -storepass $SSLPASS -keypass $SSLPASS -dname "CN=peter-kafka01.foo.bar" -storetype pkcs12
# openssl req -new -x509 -keyout ca-key -out ca-cert -days 365 -subj "/CN=foo.bar" -nodes

// 생성한 자체 서명된 CA인증서를 클라이언트가 신뢰할 수 있도록 트러스트 스토어에 추가
# keytool -keystore kafka.server.truststore.jks -alias CARoot -importcert -file ca-cert -storepass $SSLPASS -keypass $SSLPASS

// keytool을 이용한 CSR(cert-file) 생성
# keytool -keystore kafka.server.keystore.jks -alias localhost -certreq -file cert-file -storepass $SSLPASS -keypass $SSLPASS

// 인증서에 서명합니다. 
# openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days 365 -CAcreateserial -passin pass:$PASSWORD

// 키스토어에 자체 서명된 CA인증서인 ca-cert와 서명된 cert-signed를 추가합니다.
# keytool -keystore kafka.server.keystore.jks -alias CARoot -importcert -file ca-cert -storepass $SSLPASS -keypass $SSLPASS
# keytool -keystore kafka.server.keystore.jks -alias localhost -importcert -file cert-signed -storepass $SSLPASS -keypass $SSLPASS

# keytool -list -v -keystore kafka.server.keystore.jks
# keytool -list -v -keystore kafka.server.truststore.jks

 

2. 나머지 브로커에 대한 SSL구성

클러스터내 나머지 브로커들인 peter-kafka02와 peter-kafka03에서도 peter-kafka01에서 구성한 작업을 동일하게 진행해야 하므로 다음절차를 따릅니다.

 

나머지 브로커에서도 디렉토리를 만들고 키스토어를 생성합니다. 

# mkdir -p $KAFKA_HOME/ssl
# export SSLPASS=peterpass

# keytool -keystore kafka.server.keystore.jks -alias localhost -keyalg RSA -validity 365 -genkey -storepass $SSLPASS -keypass $SSLPASS -dname "CN=peter-kafka02.foo.bar" -storetype pkcs12

 

peter-kafka01에서 생성한 CA 인증서와 키 파일을 복사합니다. 여기서는 파일 복사를 위해 scp를 이용해야 하므로 ssh 키를 미리 등록해야 합니다. peter-kafka01 서버에 접속한 후 다음과 같이 ssh키를 생성하는 ssh-keygen 명령어를 실행합니다. 

# ssh-keygen
# cat .ssh/id_rsa.pub

 

peter-kafka02에서 다음 명령어를 실행하여 위 cat명령어로 출력된 해시값을 붙여넣습니다.

# vi /home/ec2-user/.ssh/authorized_keys

 

scp명령어를 이용해 peter-kafka01에서 peter-kafka02 으로 복사합니다. 이때 트러스트스토어도 함께 복사합니다. 앞선 peter-kafka01과 달리 트러스트 스토어를 추가로 생성하지는 않습니다. 모든 인증서는 자신의 유효성을 검증하기 위한 인증서 체인이 함께 제공되어야 합니다. 

# scp ca-cert peter-kafka02.foo.bar:~
# scp ca-key peter-kafka02.foo.bar:~
# scp kafka.server.truststore.jks peter-kafka02.foo.bar:~

 

peter-kafka02에서 다음과 같이 실행합니다.

# mv * $KAFKA_HOME/ssl/
# cd $KAFKA_HOME/ssl/

 

그리고 keytool을 활용하여 CSR(cert-file)을 만듭니다. 그 CSR로 서버에서 사용할 인증서(cert-signed)를 생성합니다. 

# keytool -keystore kafka.server.keystore.jks -alias localhost -certreq -file cert-file -storepass $SSLPASS -keypass $SSLPASS
# openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days 365 -CAcreateserial -passin pass:$PASSWORD

 

그 다음으로 키스토어에 CA 인증서, 브로커 서버 인증서 추가를 진행합니다. 인증서 추가 작업 시 "이 인증서를 신뢰합니까?" 라는 입력창이 나타나는데 y를 입력하면 됩니다.

# keytool -keystore kafka.server.keystore.jks -alias CARoot -importcert -file ca-cert -storepass $SSLPASS -keypass $SSLPASS
# keytool -keystore kafka.server.keystore.jks -alias localhost -importcert -file cert-signed -storepass $SSLPASS -keypass $SSLPASS

 

3. 브로커 설정에 SSL 추가

# vi $KAFKA_HOME/config/server.properties

listeners=PLANTEXT://0.0.0.0:9092, SSL:..0.0.0.0:9093
advertised.listeners=PLAINTEXT://peter-kafka01.foo.bar:9092,SSL://peter-kafka01.foo.bar:9093

ssl.truststore.location=$KAFKA_HOME/ssl/kafka.server.truststore.jks
ssl.truststore.password=peterpass
ssl.keystore.location=$KAFKA_HOME/ssl/kafka.server.keystore.jks
ssl.keystore.password=peterpass
ssl.key.password=peterpass
security.inter.broker.protocol=SSL

# systemctl restart kafka-server

 

4. 이제 클라이언트에서 브로커에 대한 트로스트 스토어를 생성하고 테스트 해보자

 

Client keystore를 생성

# $KAFKA_HOME/bin/kafka-topics.sh --bootstrap-server peter-kafka01.foo.bar:9092 --create --topic peter-test07 --partitions 1

 

HOME 경로에 ssl.config 생성

# vi ~/ssl.config

-----------------------------------------------------------------------------
security.protocol=SSL
ssl.truststore.location=$KAFKA_HOME/ssl/kafka.client.truststore.jks
ssl.truststore.password=peterpass

 

위 SSL config로 브로커에 pub sub 테스트

# KAFKA_HOME/bin/kafka-console-producer.sh --bootstrap-server peter-kafka01.foo.bar:9093 --topic peter-test07 --producer.config ~/ssl.config

# KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server peter-kafka01.foo.bar:9093 --topic peter-test07 --from-beginning --consumer.config ~/ssl.config