Streams 보안
Streams 보안
Kafka Streams 애플리케이션도 결국 카프카 클라이언트라서, 카프카의 보안 기능을 그대로 물려받아요. 이 페이지에서는 카프카 스트림즈 앱에 인증과 SSL 암호화를 적용하는 법, 그리고 보안이 적용된 클러스터에서 내부 토픽을 만들기 위해 필요한 ACL 설정을 정리해드릴게요.
출처: 문서
본문
Kafka Streams는 카프카의 보안 기능과 네이티브하게 통합되며 카프카의 모든 클라이언트 측 보안 기능을 지원해요. Streams는 Java Producer와 Consumer API를 활용해요.
스트림 처리 애플리케이션을 보호하려면 해당 카프카 프로듀서·컨슈머 클라이언트에서 보안 설정을 구성하고, Kafka Streams 애플리케이션에서 해당 구성 설정을 지정하면 돼요.
카프카는 클러스터 암호화와 인증을 지원하며, 인증된/인증되지 않은, 암호화된/암호화되지 않은 클라이언트의 혼합을 포함해요. 보안 사용은 선택 사항이에요.
다음은 관련된 몇 가지 클라이언트 측 보안 기능이에요:
- 전송 중 데이터 암호화 — 애플리케이션과 카프카 브로커 사이의 클라이언트-서버 통신 암호화를 활성화할 수 있어요. 예를 들어, 카프카에서 데이터를 읽고 쓸 때 항상 암호화를 사용하도록 애플리케이션을 구성할 수 있어요. 이것은 내부 네트워크, 공개 인터넷, 파트너 네트워크 같은 보안 도메인을 넘어 데이터를 읽고 쓸 때 중요해요.
- 클라이언트 인증 — 애플리케이션에서 카프카 브로커로의 연결에 대한 클라이언트 인증을 활성화할 수 있어요. 예를 들어, 특정 애플리케이션만 카프카 클러스터에 연결하도록 허용할 수 있어요.
- 클라이언트 인가 — 애플리케이션의 읽기·쓰기 연산에 대한 클라이언트 인가를 활성화할 수 있어요. 예를 들어, 특정 애플리케이션만 카프카 토픽에서 읽도록 허용할 수 있어요. 또한 데이터 오염이나 사기 행위를 방지하기 위해 카프카 토픽에 대한 쓰기 접근을 제한할 수도 있어요.
Apache Kafka의 보안 기능에 대한 자세한 내용은 카프카 보안을 참고해요.
보안 카프카 클러스터에 필요한 ACL 설정
카프카 클러스터는 ACL을 사용해 리소스(토픽 생성 능력 같은)에 대한 접근을 제어할 수 있어요. 그러한 클러스터에서는 Kafka Streams를 포함한 각 클라이언트가 적절한 권한으로 인가받기 위해 특정 사용자로 인증해야 해요. 특히 Streams 애플리케이션이 보안 카프카 클러스터에 대해 실행될 때, 애플리케이션을 실행하는 주체(principal)는 애플리케이션이 내부 토픽을 생성·읽기·쓰기 할 권한을 가지도록 ACL이 설정되어 있어야 해요.
group.protocol=streams를 설정해 스트림즈 리밸런스 프로토콜이 활성화되면, 토픽과 그룹 리소스에 다음 ACL이 필요해요:
| API 프로토콜 | 연산 | 리소스 | 참고 |
|---|---|---|---|
| STREAMS_GROUP_HEARTBEAT | Read | Group | 애플리케이션의 스트림즈 그룹에 필요 |
| STREAMS_GROUP_HEARTBEAT | Create | Cluster 또는 Topic | 내부 토픽을 자동 생성하는 경우에만 필요. Cluster 리소스에 Create, 또는 StateChangelogTopics와 RepartitionSourceTopics의 모든 토픽에 Create. 내부 토픽이 사전 생성된 경우 필요 없음 |
| STREAMS_GROUP_HEARTBEAT | Describe | Topic | 처음 조인할 때 애플리케이션의 토폴로지에 사용되는 모든 토픽에 필요 |
| STREAMS_GROUP_DESCRIBE | Describe | Group | 애플리케이션의 스트림즈 그룹에 필요 |
| STREAMS_GROUP_DESCRIBE | Describe | Topic | 그룹의 토폴로지에 사용되는 모든 토픽에 필요 |
앞서 언급했듯이, Kafka Streams 애플리케이션은 보안 카프카 클러스터에 대해 실행할 때 내부 토픽을 만들 수 있는 적절한 ACL이 필요해요. 이 권한을 애플리케이션에 제공하지 않으려면, 필요한 내부 토픽을 수동으로 만들면 돼요. 내부 토픽이 존재하면 Kafka Streams는 그것을 재생성하려 하지 않아요. 내부 리파티션(repartition)과 체인지로그(changelog) 토픽은 올바른 파티션 수로 생성되어야 한다는 점을 유의해요. 그렇지 않으면 Kafka Streams가 시작 시 실패해요. 토픽은 입력 토픽과 같은 파티션 수로 생성되어야 해요. 여러 토픽이 있다면 모든 입력 토픽 전체의 최대 파티션 수로요. 또한 체인지로그 토픽은 로그 컴팩션이 활성화된 상태로 생성되어야 해요. 그렇지 않으면 애플리케이션이 데이터를 잃을 수 있어요. 윈도우 KTable용 체인지로그 토픽은 "delete,compact"를 적용하고 해당 저장소 리텐션 시간을 기준으로 리텐션 시간을 설정해요. 조기 삭제를 피하려면 저장소 리텐션 시간에 델타를 더하세요. 기본적으로 Kafka Streams는 저장소 리텐션 시간에 24시간을 더해요. 필요한 내부 토픽의 이름에 대한 자세한 내용은 Topology#describe()를 통해 알 수 있어요. 모든 내부 토픽은 <application.id>-<operatorName>-<suffix> 명명 패턴을 따르며 suffix는 repartition 또는 changelog예요. 향후 릴리스에서 이 명명 패턴은 보장되지 않으며 공개 API의 일부가 아니라는 점을 유의해요.
모든 내부 토픽과 임베드된 컨슈머 그룹 이름이 application id로 접두사가 붙으므로, 접두사가 붙은 리소스 패턴에 ACL을 사용해 --resource-pattern-type prefixed --topic your.application.id --operation All처럼 이 접두사로 시작하는 모든 토픽과 컨슈머 그룹을 클라이언트가 관리하도록 허용하는 접근 제어 목록을 구성하는 것을 권장해요 (자세한 내용은 KIP-277과 KIP-290 참조).
보안 예시
목적은 카프카 클러스터와 통신할 때 클라이언트 인증을 활성화하고 전송 중 데이터를 암호화하도록 Kafka Streams 애플리케이션을 구성하는 것이에요.
이 예시는 클러스터의 카프카 브로커에 이미 보안 설정이 있고, 필요한 SSL 인증서가 로컬 파일시스템 위치에 애플리케이션이 접근할 수 있게 있다고 가정해요. 예를 들어 Docker를 사용한다면 이러한 SSL 인증서를 Docker 이미지의 올바른 위치에도 포함해야 해요.
다음 스니펫은 Kafka Streams 애플리케이션과 그것이 읽고 쓰는 카프카 클러스터 사이의 전송 중 데이터에 클라이언트 인증과 SSL 암호화를 활성화하는 설정을 보여줘요:
# Essential security settings to enable client authentication and SSL encryption
bootstrap.servers=kafka.example.com:9093
security.protocol=SSL
ssl.truststore.location=/etc/security/tls/kafka.client.truststore.jks
ssl.truststore.password=test1234
ssl.keystore.location=/etc/security/tls/kafka.client.keystore.jks
ssl.keystore.password=test1234
ssl.key.password=test1234
이 설정을 애플리케이션의 Properties 인스턴스에 구성하세요. 이 설정은 카프카에서 읽거나 쓰는 전송 중 데이터를 암호화하고, 애플리케이션이 통신하는 카프카 브로커에 대해 자신을 인증할 거예요. 이 예시는 클라이언트 인가를 다루지 않는다는 점을 유의해요.
// Code of your Java application that uses the Kafka Streams library
Properties settings = new Properties();
settings.put(StreamsConfig.APPLICATION_ID_CONFIG, "secure-kafka-streams-app");
// Where to find secure Kafka brokers. Here, it's on port 9093.
settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka.example.com:9093");
//
// ...further non-security related settings may follow here...
//
// Security settings.
// 1. These settings must match the security settings of the secure Kafka cluster.
// 2. The SSL trust store and key store files must be locally accessible to the application.
settings.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
settings.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "/etc/security/tls/kafka.client.truststore.jks");
settings.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "test1234");
settings.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, "/etc/security/tls/kafka.client.keystore.jks");
settings.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, "test1234");
settings.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, "test1234");
애플리케이션에서 보안 설정을 잘못 구성하면, 보통 시작 직후 런타임에 실패해요. 예를 들어 ssl.keystore.password 설정에 잘못된 비밀번호를 입력하면 다음과 유사한 오류 메시지가 기록되고 애플리케이션이 종료돼요:
# Misconfigured ssl.keystore.password
Exception in thread "main" org.apache.kafka.common.KafkaException: Failed to construct kafka producer
[...snip...]
Caused by: org.apache.kafka.common.KafkaException: org.apache.kafka.common.KafkaException:
java.io.IOException: Keystore was tampered with, or password was incorrect
[...snip...]
Caused by: java.security.UnrecoverableKeyException: Password verification failed
잘못 구성된 애플리케이션을 빨리 찾으려면 Kafka Streams 애플리케이션 로그 파일에서 이러한 오류 메시지를 모니터링하세요.