Docker compose로 올린 Kafka에 ACL, SASL/PLAIN 인증 설정하기

 

들어가며

기본적으로 카프카는 어느 누구나 클러스터에 접근하여 메시지를 생성하거나 읽을 수 있다.

이로인해 보안이 적용되지 않은 상태에서는 데이터의 무단 액세스와 무단 조작이 가능하므로,

중요한 정보를 다루는 시스템에서는 보안을 강화하는 것이 중요하다.

 

그러므로 카프카에 누가 요청을 보내고, 그 요청자의 권한에 따라서 요청 범위를 제한하는

카프카 인증, 인가에 대해서 알아보고 직접 적용해보고자 한다.

 

 

 개념 

카프카 인증

카프카에서 인증을 처리하는 방식으로 SSL 방식과 SASL 방식으로, 크게 두 가지 방식이 존재한다.

• SSL : 인증서를 활용한 인증 방식
  • 데이터를 암호화하고 통신 채널을 안전하게 유지하여 중간에서의 데이터 탈취나 변조를 방지한다.

• SASL : SASL 프로토콜을 이용한 인증 방식
  • 클라이언트가 카프카 브로커에 대해 인증하는 데 사용된다. 즉, 클라이언트가 자신의 신원을 확인하고 시스템에 로그인할 수 있게 하는 작업이다.

 

SSL(Secure Sockets Layer)을 적용할 경우, 클라이언트에서 메시지가 암호화되어 전달된다.

이 암호화된 메시지는 브로커에서 다시 복호화 되는데, 이때 CPU 부하가 발생할 수 있다.

현재는 더 안전한 후속버전인 TLS (Transport Layer Security) 프로토콜로 대체되었지만,

여전히 SSL라는 용어가 널리 사용되고 있다.

 

SASL(Simple Authentication and Security Layer) 방식은 사용자가 카프카 클러스터에 접속하고자 할 때, 

SASL을 사용하여 인증 처리를 위한 별도의 층(Layer)을 분리하고, 그 층에서 사용자 인증을 처리한다.

따라서 처음에 인증 단계를 거치면 이후에 애플리케이션 단에서 인증을 처리하지 않아도 된다.

 

사실상 SSL과 SASL을 모두 적용하는 것이 데이터의 기밀성, 무결성, 인증을 강화하여 더 안전한 시스템을

만들 수 있다.

(이번에는 SASL와 ACL만 다루고, SSL은 추후에 다뤄볼 예정이다.)

 

 

카프카에서 지원하는 SASL 메커니즘

 

추상화된 SASL 프레임워크를 통한 인증 처리 과정

 

위 그림에서 볼 수 있는 것처럼 SASL 프레임워크는 다양한 메커니즘으로 인증을 처리할 수 있다.
카프카에서 기본적으로 제공하는 SASL 메커니즘은 다음 4가지가 존재한다.

• PLAIN
  • 평문으로 아이디, 패스워드를 설정하여 인증을 하는 가장 기본적이고 고전적인 방식

 

• SCRAM
  
  • PBKDF2 암호화 알고리즘을 활용해 생성된 해시를 활용하며
  • 비밀번호에 솔트(salt)를 추가하여 보안을 높인 방식

 

• OAUTHBEARER
  • OAuth2를 기반으로 하는 인증 방식으로, 클라이언트가 OAuth2 토큰을 사용하여 서버에 인증한다.
  • 기본적으로 카프카에서 제공하는 oauthbearer callbackhander는 아무런 보안 처리가 되어있지 않기 때문에, 별도의 OAuth 인증 서버와 별도의 OAuth 인증을 처리할 CallbackHandler를 구현해야 한다.

 

• GSSAPI (KERBEROS)
  • 노드간 통신에서 보안을 클라이언트가 티켓을 발급받아 본인의 신원을 증명하면 인증되는 매우 안전한 방법
  • 별도의 인증 및 티켓검증용 서버가 필요하며 서버가 불능이 될 경우 인증 불가하므로 관리에 주의해야 한다.

 

 

카프카 인가

인증이 완료되면, 클라이언트 계정 정보를 바탕으로 권한을 확인한다.

카프카는 기본적으로 ACL(Access Control List) 방식으로 권한을 확인하는데,

권한 정보를 주키퍼에 저장하고 관리하므로써 인가를 처리한다.

 

권한 관리가 되는 카프카 리소스 목록

• 토픽 (Topic)
• 컨슈머 그룹 (Group)
• 클러스터 (Cluster)
• 트랜젝션 ID (TransactionalId)
• Delegation Token

 

리소스에 따른 해당 작업 권한 목록

• 읽기 (Read)
• 쓰기 (Write)
• 생성 (Create)
• 삭제 (Delete)
• 수정 (Alter)
• 상세 (Describe)
• ClusterAction
• DescribeConfigs
• AlterConfigs
• IdempotentWrite
• All

 

 

 

 구현 

1. docker-compose.yml 수정하기

Zookeeper 컨테이너와 Kafka 컨테이너 각각의 환경 변수에 기존 내용에 추가하여 SASL 관련 설정을 넣어준다.

version: '3.7'
services:
  zookeeper:
    image: wurstmeister/zookeeper:3.4.6
    environment:
      JVMFLAGS: "-Djava.security.auth.login.config=/opt/zookeeper-3.4.6/conf/zookeeper_jaas.conf -Dzookeeper.authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider -Dzookeeper.requireClientAuthScheme=sasl"
    volumes:
      - ./zookeeper_jaas.conf:/opt/zookeeper-3.4.6/conf/zookeeper_jaas.conf
    container_name: zookeeper
    ports:
      - "2181:2181"
  kafka:
    image: wurstmeister/kafka:2.13-2.8.1
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_HOST_NAME: {Public IP}
      KAFKA_SECURITY_INTER_BROKER_PROTOCOL: SASL_PLAINTEXT
      KAFKA_LISTENERS: SASL_PLAINTEXT://:9092
      KAFKA_ADVERTISED_LISTENERS: SASL_PLAINTEXT://{Public IP}:9092
      KAFKA_SASL_ENABLED_MECHANISMS: PLAIN
      KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: PLAIN
      KAFKA_OPTS: "-Djava.security.auth.login.config=/opt/kafka/config/kafka_jaas.conf"
      KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "true"
      KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.auth.SimpleAclAuthorizer
      KAFKA_SUPER_USERS: User:{Super Users Username}
    volumes:
      - /var/run/docker.sock
      - ./kafka_jaas.conf:/opt/kafka/config/kafka_jaas.conf
    depends_on:
      - zookeeper

 

 

 

 

2. 인증 정보 conf 파일 구성

docker-compose.yml의 volumes 부분을 보다시피

zookeeper_jaas.conf와 kafka_jaas.conf 파일이 존재하는 각 로컬 디렉토리를 해당 컨테이너 경로에 마운트 시키므로

각 로컬 디렉토리 경로에 zookeeper_jaas.conf와 kafka_jaas.conf 파일을 작성해준다.

(나의 경우 두 파일이 저장된 경로가 같다)

 

zookeeper_jaas.conf

Server {
    org.apache.zookeeper.server.auth.DigestLoginModule required
    username="admin"
    password="1234"
    user_admin="1234"
    user_test="5678";
};

• username, password : Super User 설정
• user_{UserName} = "{Password}"

 

참고로 docker-compose로 올렸기 때문에, 아래와 같은 모듈을 사용한다.

kafka.common으로 설정하면 해당 모듈이 없다고 에러가 발생하게 된다.

(X) org.apache.kafka.common.security.plain.PlainLoginModule required

(O) org.apache.zookeeper.server.auth.DigestLoginModule required

 

kafka_jaas.conf

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="1234"
    user_admin="1234"
    user_test="5678";
};
Client {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="1234";
};

• KafkaServer : zookeeper_jaas.conf 와 똑같이 Super 유저와 일반 유저 설정한다.
• Client : Zookeeper Server에 접속할 유저 정보

 

 

 

3. docker-compose 실행하기 (Zookeeper, Kafka 실행)

$ docker-compose up -d

 

 

이제 잘 띄워졌나 확인해보자

$ docker ps -a

Docker에서 실행 중인 Zookeeper와 Kafka 컨테이너가 성공적으로 실행중인 상태이다.

 

docker logs {CONTAINER ID}로 Zookeeper와 Kafka끼리 SASL 연결에 있어서

각 컨테이너에서 성공했을 때의 로그를 살펴보자

 

Zookeeper

2024-01-18 02:25:57,959 [myid:] - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:SaslServerCallbackHandler@118] - Successfully authenticated client: authenticationID=admin;  authorizationID=admin.
2024-01-18 02:25:58,009 [myid:] - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:SaslServerCallbackHandler@134] - Setting authorizedID: admin
2024-01-18 02:25:58,010 [myid:] - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:ZooKeeperServer@964] - adding SASL authorization for authorizationID: admin

 

 

Kafka

굉장히 많은데 일부분만 (listener, sasl)

[2024-01-18 02:25:58,459] INFO KafkaConfig values:
        advertised.host.name = {Public IP}
        advertised.listeners = SASL_PLAINTEXT://{Public IP}:9092
        advertised.port = null
				...
				listener.security.protocol.map = PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
        listeners = SASL_PLAINTEXT://:9092
				...
				sasl.client.callback.handler.class = null
        sasl.enabled.mechanisms = [PLAIN]
        sasl.jaas.config = null
        sasl.kerberos.kinit.cmd = /usr/bin/kinit
        sasl.kerberos.min.time.before.relogin = 60000
        sasl.kerberos.principal.to.local.rules = [DEFAULT]
        sasl.kerberos.service.name = null
        sasl.kerberos.ticket.renew.jitter = 0.05
        sasl.kerberos.ticket.renew.window.factor = 0.8
        sasl.login.callback.handler.class = null
        sasl.login.class = null
        sasl.login.refresh.buffer.seconds = 300
        sasl.login.refresh.min.period.seconds = 60
        sasl.login.refresh.window.factor = 0.8
        sasl.login.refresh.window.jitter = 0.05
        sasl.mechanism.controller.protocol = GSSAPI
        sasl.mechanism.inter.broker.protocol = PLAIN
        sasl.server.callback.handler.class = null
        security.inter.broker.protocol = SASL_PLAINTEXT

 

[2024-01-18 02:25:57,940] INFO [ZooKeeperClient Kafka server] Connected. (kafka.zookeeper.ZooKeeperClient)

 

 

 

4. 유저에 대한 계정 권한 부여 방법

우선 카프카 컨테이너의 bash로 접근한다

$ docker exec -it kafka bash

 

권한 부여는 아래 경로에 위치한 kafka-acls.sh를 통해 수행된다.

$ cd opt/kafka/bin

 

/opt/kafka/bin 경로 안에 있는 kafka-acls.sh로 권한 부여를 수행한다.

 

Kafka Single로 구축했을 때

(나의 경우 Single 명령어로 수행했음)

./kafka-acls.sh --authorizer-properties zookeeper.connect=zookeeper:2181 --add --allow-principal User:test --operation All --topic '*' --group '*'

 

Kafka Cluster로 구축했을 때

뒤에 --cluster를 추가해준다.

./kafka-acls.sh --authorizer-properties zookeeper.connect=zookeeper:2181 --add --allow-principal User:test --operation All --topic '*' --group '*' --cluster

• -allow-principal User:{UserName} : 권한 부여할 유저
• -operation : producer, consumer, All 권한
• -topic : topic, '*'
• -group : group Id (producer만 지정할 땐 필요X)

 

권한 생성 후 콘솔 메시지

 

유저 권한 삭제

권한 삭제는 --add 대신 --remove로 변경해주고 원하는 권한을 설정해주면 된다.

kafka-acls.sh --authorizer-properties zookeeper.connect=zookeeper:2181 --add --allow-principal User:test --operation All --topic '*' --group '*'

 

 

 

5. Producer / Consumer SASL 적용 

해당 작업과 실행은 /opt/kafka/config 경로에서 수행했다.
(로컬 경로에 미리 만들고 docker-compose.yml 작성 시, 마운트 시켜줘도 상관없음)

 

producer_jaas.conf

bootstrap.servers=SASL_PLAINTEXT://{Public IP}:9092
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
username="admin" \
password="1234";

 

consumer_jaas.conf

client.id=admin

bootstrap.servers=SASL_PLAINTEXT://{Public IP}:9092
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
username="admin" \
password="1234";

 

 

producer와 consumer를 실행하기 전에,

토픽을 생성하지 않았다면 생성을 한 후, 실행을 진행하자

 

console에서 producer 실행

kafka-console-producer.sh --broker-list {Public IP}:9092 --topic {Topic} --producer.config=../config/producer_jaas.conf

 

console에서 consumer 실행

kafka-console-consumer.sh --bootstrap-server {Public IP}:9092 --topic {Topic} --consumer.config=../config/consumer_jaas.conf --from-beginning

 

--from-beginning은 기호에 맞게 선택

예를 들어, 위의 명령어를 사용하여 특정 토픽의 메시지를 소비할 때 --from-beginning 옵션을 추가하면,

해당 토픽에 존재하는 모든 이전 메시지들을 소비하면서

현재 시점 이후에도 새로운 메시지를 계속 수신할 수 있게 된다.

 

실행 성공 화면

왼쪽 Producer에서 메시지를 입력하면, 오른쪽 Consumer에서 잘 꺼내오고 있음

 

허가되지 않은 User로 console에서 Producer / Consumer로 접근 시

disconnected 되면서 에러 메시지가 발생한다.

[2024-01-18 07:33:02,679] INFO [SocketServer listenerType=ZK_BROKER, nodeId=1001] Failed authentication with /{Public IP} (Unexpected Kafka request of type METADATA during SASL handshake.) (org.apache.kafka.common.network.Selector)

 

 

 

 

콘솔에서 SASL 인증이 잘 적용된 것을 확인했으니,

다음 시간에는 스프링부트와 Kafka 간의 연동을 위해 스프링부트에서 SASL 설정을 다뤄보고자 한다.

 

 

 

끄으으으읕

 

 

 

Reference

https://kafka.apache.org/documentation/#security

Apache Kafka

 

https://always-kimkim.tistory.com/entry/kafka101-security

[Kafka 101] 카프카 보안 - 인증과 인가 (Kafka Security - Authentication & Authorization)

 

https://velog.io/@limsubin/Kafka-SASLPALIN-인증-설정을-해보자#3-구성-전-작업

Kafka acl SASL/PALIN 인증 설정을 해보자!

 

https://heodolf.tistory.com/16#google_vignette

[KAFKA] 무작정 시작하기 (5) - SASL 인증 설정

 

https://stackoverflow.com/questions/61924877/sasl-authentication-in-docker-zookeeper-and-kafka

SASL authentication in docker zookeeper and kafka

 

https://stackoverflow.com/questions/58833922/kafka-not-starting-up-if-zookeeper-set-acl-is-set-to-true

Kafka not starting up if zookeeper.set.acl is set to true

 

https://stackoverflow.com/questions/40664265/who-is-setting-up-the-authorizations-for-kafka-cluster

Who is setting up the authorizations for kafka cluster

 

https://stackoverflow.com/questions/43559328/got-user-level-keeperexception-when-processing

Got user-level KeeperException when processing

 

https://github.com/factorhouse/kafka-local/blob/main/docker-compose.yml