[실전 카프카 개발부터 운영까지] 9장. 카프카 보안

[youngpark17]

9장 카프카 보안

카프카 보안의 3요소

• 암호화: 웹에 비유하자면 http x -> https
• 인증: id/pw
• 권한: 클라이언트의 권한을 분리

암호화

• 암호화는 주로 SSL을 이용

  ![image](

인증(SASL-Simple Authentication and Security Layer)

• SASL은 인터넷 프로토콜에서 인증과 데이터 보안을 위한 프레임워크로, 카프카에서도 사용됩니다. 카프카에서는 네가지 SASL 메커니즘을 지원합니다.
• GSSAPI: 커버로스 인증방식으로 많이 사용되는 인증 방식, 회사에 커버로스 서버가 있다면 권장됨. 하나의 렐름(설정)으로 모든 애플리케이션 적용하는 것이 좋음
• PLAIN: 아이디와 비밀번호를 텍스트 형태로 사용, 개발 환경에서 테스트 등의 목적으로 권장
• SCRAM-SHA-256, SCRAM-SHA-512(Salted Challenge Response Authentication Mechanism), 본래의 암호에 해시된 내용을 추가 하여 암호가 유출되어도 본래 암호를 알 수 없다. 커버로스 서버가 없다면 가장 좋은 방식
• OAUTHBEARER: 2.0버전부터 지원되었지만, 현재 카프카에서 제공하는기능은 배우 한정적. 운영에서는 별도의 핸들러 구성 필요

권한(ACL)

• 프로듀서에서 잘못된 토픽으로 메시지를 발행한다면 해당 토픽을 구독하는 컨슈머에서 파싱에러등으로 인해 장애가 발생할 수 있음
• 각 클라이언트별로 올바른 권한을 할당하여 이슈 방지
• 간단히 CLI로 ACL추가 가능하며 ACL 규칙은 주키퍼에 저장됨, 리소스 타입(토픽, 그룹, 클러스터, 트랜잭셔널 ID, 위임 토큰)등으로 구체적 설정 가능

SSL을 이용한 카프카 암호화

• 자바기반 애플리케이션에서는 키스토어라고 불리는 인터페이스를 통해 퍼블릭 키, 프라이빗 키, 인증서를 추상화해 제공. 카프카 또한 마찬가지

브로커 키스토어 생성

• 클러스터 내 각 브로커마다 프라이빗 키와 인증서를 만들어야 함.
• 클라이언트의 트러스트 스토어와 서버의 키스토어는 자바 애플리케이션을 이용해 SSL 연결을 할 때 사용함. 키스토어와 트러스트스토어 모두 keytool을 이용해 관리됨
• 키스토어는 서버측면에서 프라이빗 키와 인증서를 저장하며, 자격 증명을 제공하고 그외에 프라이빗하고 민감한 정보를 저장
• 트러스트스토어는 클라이언트 측면에서 서버가 제공하는 인증서를 검증하기 위한 퍼블릭 키와 서버와 SSL 연결에서 유효성을 검사하는 서명된 인증서를 저장하며, 민감한 정보는 저장하지 않습니다.

CA 인증서 생성

• 일반적으로는 인증기관에서 인증서 받아야하지만 실습에서는 자체 서명된 CA 인증서를 생성하여 사설 인증서에 자체 서명된 CA 서명을 통해 신뢰할 수 있는 내부 환경 구축

트러스트스토어 생성

• 자체 서명된 CA 인증서를 클라이언트가 신뢰 할 수 있도록 트러스트스토어에 추가합니다.

인증서 서명

• 키스토어에서 인증서 추출해 자체서명

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

브로커 설정에 SSL 추가

SSL 기반 메시지 전송

커버로스(SASL)를 이용한 카프카 인증

• 커버로스는 티켓을 기반으로 하는 컴퓨터 네트워크 인증 프로토콜, 사용자의 신원을 식별하기 위한 용도로 사용. 하나의 커버로스 티켓을 이용해 여러 서비스에서 같이 적용할 수 있는 싱글 사인온(SSO)을 지원하기 때문
• 커버로스 위키
• 커버로스 프로토콜
• 커버로스 유튜브

[youngpark17]

sudo mkdir -p /usr/local/kafka/ssl
cd /usr/local/kafka/ssl
export SSLPASS=peterpass

# 키스토어 생성
sudo 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

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

## ca인증서 생성
sudo openssl req -new -x509 -keyout ca-key -out ca-cert -days 356 -subj "/CN=foo.bar" -nodes

## 트러스트스토어 생성
sudo keytool -keystore kafka.server.truststore.jks -alias CARoot -importcert -file ca-cert -storepass $SSLPASS -keypass $SSLPASS
keytool -list -v -keystore kafka.server.truststore.jks

## 인증서 서명
sudo keytool -keystore kafka.server.keystore.jks -alias localhost -certreq -file cert-file -storepass $SSLPASS -keypass $SSLPASS

sudo openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days 365 -CAcreateserial -passin pass:$PASSWORD ## 확인필요 SSLPASS로..?

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

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

ssh-keygen

cat .ssh/id_rsa.pub

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

cd /usr/local/kafka/ssl/

### scp로 브로커 2,3에 트러스트스토어 복사
scp ca-cert peter-kafka02.foo.bar:~
scp ca-key peter-kafka02.foo.bar:~
scp kafka.server.truststore.jks peter-kafka02.foo.bar:~

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

peter-kafka02 서버

sudo mv * /usr/local/kafka/ssl
cd /usr/local/kafka/ssl/

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

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

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

peter-kafka03 서버

sudo mv * /usr/local/kafka/ssl/
cd /usr/local/kafka/ssl

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

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

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

브로커간의 통신은 plaintext 브로커 1,2,3 작업필요

sudo vi /usr/local/kafka/config/server.properties

listeners=PLAINTEXT://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=/usr/local/kafka/ssl/kafka.server.truststore.jks
ssl.truststore.password=peterpass
ssl.keystore.location=/usr/local/kafka/ssl/kafka.server.keystore.jks
ssl.keystore.password=peterpass
ssl.key.password=peterpass
security.inter.broker.protocol=SSL

sudo systemctl restart kafka-server

openssl s_client -connect peter-kafka01.foo.bar:9093 -tls1 </dev/null 2>/dev/null | grep -E 'Verify return code'

SSL 기반 메시지 전송

cd /usr/local/kafka/ssl/
export SSLPASS=peterpass
sudo keytool -keystore kafka.client.truststore.jks -alias CARoot -importcert -file ca-cert -storepass $SSLPASS -keypass $SSLPASS

/usr/local/kafka/bin/kafka-topics.sh --bootstrap-server peter-kafka01.foo.bar:9092 --create --topic peter-test07 --partitions 1 --replication-factor 3

vi /home/ec2-user/ssl.config

security.protocol=SSL
ssl.truststore.location=/usr/local/kafka/ssl/kafka.client.truststore.jks
ssl.truststore.password=peterpass

/usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties

커버로스 인증

cd ansible_playbook
ansible-playbook -i hosts kerberos.yml

sudo kadmin.local -qa "add_principal -randkey peter01@F00.BAR"
sudo kadmin.local -qa "add_principal -randkey peter02@F00.BAR"
sudo kadmin.local -q "add_principal -randkey admin@FOO.BAR"

sudo kadmin.local -q "add_principal -randkey kafka/peter-kafka01.foo.bar@FOO.BAR"
sudo kadmin.local -q "add_principal -randkey kafka/peter-kafka02.foo.bar@FOO.BAR"
sudo kadmin.local -q "add_principal -randkey kafka/peter-kafka03.foo.bar@FOO.BAR"

mkdir -p /home/ec2-user/keytabs/
sudo kadmin.local -q "ktadd -k /home/ec2-user/keytabs/peter01.user.keytab peter01@FOO.BAR"
sudo kadmin.local -q "ktadd -k /home/ec2-user/keytabs/peter02.user.keytab peter02@FOO.BAR"
sudo kadmin.local -q "ktadd -k /home/ec2-user/keytabs/admin.user.keytab admin@FOO.BAR"
sudo kadmin.local -q "ktadd -k /home/ec2-user/keytabs/peter-kafka01.service.keytab kafka/peter-kafka01.foo.bar@FOO.BAR"
sudo kadmin.local -q "ktadd -k /home/ec2-user/keytabs/peter-kafka02.service.keytab kafka/peter-kafka02.foo.bar@FOO.BAR"
sudo kadmin.local -q "ktadd -k /home/ec2-user/keytabs/peter-kafka03.service.keytab kafka/peter-kafka03.foo.bar@FOO.BAR"

sudo chown -R ec2-user.ec2-user keytabs

scp -i keypair.pem -r peter-zk01.foo.bar:~/keytabs /home/ec2-user
sudo mv keytabs /usr/loca/kafka

cat /etc/krb5.conf

kinit -kt /usr/local/kafka/keytabs/peter01.user.keytab peter01
klist

kinit -kt /usr/local/kafka/keytabs/peter-kafka01.service.keytab kafka/peter-kafka01.foo.bar

sudo vi/usr/local/kafka/config/server.properties
SASL_PLAINTEXT://0.0.0.0:9094
SASL_PLAINTEXT://peter-kafka01.foo.bar:9094

security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=GSSAPI
sasl.enabled.mechanism=GSSAIP
sasl.kerberos.service.name=kafka

sudo vi /usr/local/kafka/config/kafka_server_jaas.conf

KafkaServer {
  com.sun.security.auth.module.Krb5LoginModule required
  useKeyTab=true
  storeKey=true
  keyTab="/usr/local/kafka/keytabs/peter-kafka01.service.keytab"
  principal="kafka/peter-kafka01.foo.bar@FOO.BAR";
};

sudo vi /usr/local/kafka/config/jmx

KAFKA_OPTS="-Djava.security.auth.login.config=/usr/local/kafka/config/kafka_server_jaas.conf"

sudo systemctl restart kafka-server

sudo netstat -ntlp | grep 9094

vi kafka_client_jaas.conf

KafkaClient {
  com.sun.security.auth.module.Krb5LoginModule requried
  useTicketCache=true;
}

sasl.mechanism=GSSAPI
security.protocol=SASL_PLAINTEXT
sasl.kerberos.service.name=kafka

kinit -kt /usr/local/kafka/keytabs/peter01.user.keytab peter01

#### 메시지 전송
/usr/local/kafka/bin/kafka-console-producer.sh --bootstrap-server peter-kafka01.foo.bar:9094 --topic peter-test08 --producer.config kerberos.config

#### 메시지 컨슈밍
/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server peter-kafka01.foo.bar:9094 --topic peter-test08 --from-beginning --consumer . config kerberos.config

#### 티켓삭제
kdestroy

#### 티켓리스트 확인(삭제확인)
klist

#### 키 삭제되었으므로 컨슘 불가
/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server peter-kafka01.foo.bar:9094 --topic peter-test08 --from-begining --consumer . config kerberos.config

acl 이용한 카프카 권한 설정

sudo vi /usr/kafka/config/server.properties

security.iter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=GSSAPI
sasl.enabled.mechanism=GSSAPI
sasl.kerberos.service.name=kafka

authorizer.class.name=kafka.security.authorizer.AclAuthorizer
super.users=User:admin;User:kafka

sudo systemctl restart kafka-server

unset KAFKA_OPTS
/usr/local/kafka/bin/kafka-topics.sh --zookeeper peter-zk01.foo.bar:2181 --create --topic peter-test09 --partitions 1 --replication-factor 1
/usr/local/kafka/bin/kafka-topics.sh --zookeeper peter-zk01.foo.bar:2181 --create --topic peter-test10 --partitions 1 --replication-factor 1

### peter01, peter02, admin 유저로 ACL 규칙 및 테스트

- peter01유저는 peter-test09 토픽에 대해 읽기 쓰기
- peter02 유저는 peter-test10 토픽에 대해 읽기 쓰기
- admin 유저는 peter-test09, peter-test10 토픽에 대해 읽기 쓰기

/usr/local/kafka/bin/kafka-acls.sh --autorizer-properties zookeeper.connect=peter-zk01.foo.bar:2181 --add --allow-principal User:peter01 --operation Read --operation Writer --operation DESCRIBE --topic peter-test09

/usr/local/kafka/bin/kafka-acls.sh --autorizer-properties zookeeper.connect=peter-zk01.foo.bar:2181 --add --allow-principal User:peter02 --operation Read --operation Writer --operation DESCRIBE --topic peter-test10

/usr/local/kafka/bin/kafka-acls.sh --autorizer-properties zookeeper.connect=peter-zk01.foo.bar:2181 --list

kinit -kt /usr/local/kafka/keytabs/peter01.user.keytab peter01

export KAFKA_OPTS="-Djava.security.auth.login.config=/home/ec2-user/kafka_client_jaas.conf"

/usr/local/kafka/bin/kafka-console-producer.sh --bootstrap-server peter-kafka01.foo.bar:9094 --topic peter-test09 --producer.config kerberos.config

/usr/local/kafka/bin/kafka-console-producer.sh --bootstrap-server peter-kafka01.foo.bar:9094 --topic peter-test10 --producer.config kerberos.config

.. 이후 유저별 컨슘 테스트