Kafka Oauth 인증 추가하기

원문 : https://medium.com/egen/how-to-configure-oauth2-authentication-for-apache-kafka-cluster-using-okta-8c60d4a85b43

How to Configure OAuth2 Authentication for Apache Kafka Cluster using Okta

 

 

우선 kafka oauth Class File을 다운로드 받는다 

 

repository에서 다운로드 : https://github.com/vishwavangari/kafka-oauth2

이걸 Build수행하고 target에 있는 kafka-oauth2-0.0.1.jar 을 내 kafka 서버에 넣자

 

필자의 경우 여기로 이동

/root/download/kafka-oauth2-master/build/libs/kafka-oauth2-0.0.1.jar

 

kafka home의 lib 경로로 이동시킴

# cp /root/download/kafka-oauth2-master/build/libs/kafka-oauth2-0.0.1.jar /root/kafka_2.12-3.2.0/libs/

 

keyCloak 혹은 okta에서 아래와 같은 Client를 정의한다. 이때 인증방식은 id/Secret 으로 하였다.

중요한것은 아래 3개의 Client 모두 동일한 Client Scope에 정의되어 있고 별도의 Rule을 매핑했다는 것이다.

• kafkabroker
• kafkaconsumerapp
• kafkaproducerapp

 

이제 kafka 서버로 가서 home/config 로 가자

 

 

아래는 내가 반영한 샘플

# vi server.properties
-----------------------------------------------------------------------------------
##########SECURITY using OAUTHBEARER authentication ###############

sasl.enabled.mechanisms=OAUTHBEARER
sasl.mechanism.inter.broker.protocol=OAUTHBEARER
security.inter.broker.protocol=SASL_PLAINTEXT

listeners=SASL_PLAINTEXT://localhost:9093
advertised.listeners=SASL_PLAINTEXT://localhost:9093

#Authorizer for ACL
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
super.users=User:kafkabroker;

################ OAuth Classes #####################

sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required OAUTH_LOGIN_SERVER=keycloak.mykeycloak-development.nip.io OAUTH_LOGIN_ENDPOINT='/realms/demo-jpkim/protocol/openid-connect/token' OAUTH_LOGIN_GRANT_TYPE=client_credentials OAUTH_LOGIN_SCOPE=broker.kafka OAUTH_AUTHORIZATION='Basic a2h6W@@@@@@@@#####WxDQlZUTk4=' OAUTH_INTROSPECT_SERVER=keycloak.mykeycloak-development.nip.io
OAUTH_INTROSPECT_ENDPOINT='/realms/demo-jpkim/protocol/openid-connect/token/introspect' OAUTH_INTROSPECT_AUTHORIZATION='Basic a@@@@@@@@#####Ed6ZEROdFh6WWxDQlZUTk4=';
listener.name.sasl_plaintext.oauthbearer.sasl.login.callback.handler.class=com.oauth2.security.oauthbearer.OAuthAuthenticateLoginCallbackHandler
listener.name.sasl_plaintext.oauthbearer.sasl.server.callback.handler.class=com.oauth2.security.oauthbearer.OAuthAuthenticateValidatorCallbackHandler

########## SECURITY using OAUTHBEARER authentication ###############

 

 

 

그리고 jaas config를 수정하자

#vi kafka_server_jaas.conf

----------------------

KafkaServer {
org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required
LoginStringClaim_sub="kafkabroker";
};

 

이제 kafka 서버를 기동시키자

# export KAFKA_OPTS="-Djava.security.auth.login.config=/root/kafka_2.12-3.2.0/config/kafka_server_jaas.conf -DOAUTH_WITH_SSL=true -DOAUTH_LOGIN_SERVER=keycloak.mykeycloak-development.nip.io -DOAUTH_LOGIN_ENDPOINT=/realms/demo-jpkim/protocol/openid-connect/token -DOAUTH_LOGIN_GRANT_TYPE=client_credentials -DOAUTH_LOGIN_SCOPE=kafka -DOAUTH_INTROSPECT_SERVER=keycloak.mykeycloak-development.nip.io -DOAUTH_INTROSPECT_ENDPOINT=/realms/demo-jpkim/protocol/openid-connect/token/introspect -DOAUTH_AUTHORIZATION=Basic%20a2Fma2Fi!!!!!@@@@@@@@@@@d6 -DOAUTH_INTROSPECT_AUTHORIZATION=Basic%20a2Fm!!!!!@@@@@@@@@@@VEd6ZEROdFh6WWxDQlZUTk4="

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

 

※ 만약  kafka.security.auth.SimpleAclAuthorizer 관련해서 오류가 발생한다면 server.properties에서 kafka.security.auth.SimpleAclAuthorizer를 kafka.security.authorizer.AclAuthorizer 로 수정해서 다시 올린다.

 

 

 

원문

# vi server.properties

##########SECURITY using OAUTHBEARER authentication ###############

sasl.enabled.mechanisms=OAUTHBEARER

sasl.mechanism.inter.broker.protocol=OAUTHBEARER

security.inter.broker.protocol=SASL_PLAINTEXT

listeners=SASL_PLAINTEXT://localhost:9093

advertised.listeners=SASL_PLAINTEXT://localhost:9093

#Authorizer for ACL

authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

super.users=User:<brokerapp-clientId>;

################ OAuth Classes #####################

sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required OAUTH_LOGIN_SERVER=<auth-server-url> OAUTH_LOGIN_ENDPOINT='/oauth2/default/v1/token' OAUTH_LOGIN_GRANT_TYPE=client_credentials OAUTH_LOGIN_SCOPE=broker.kafka OAUTH_AUTHORIZATION='Basic <encoded-clientId:clientsecret>' OAUTH_INTROSPECT_SERVER=<auth-server-url> OAUTH_INTROSPECT_ENDPOINT='/oauth2/default/v1/introspect' OAUTH_INTROSPECT_AUTHORIZATION='Basic <encoded-clientId:clientsecret>';

listener.name.sasl_plaintext.oauthbearer.sasl.login.callback.handler.class=com.oauth2.security.oauthbearer.OAuthAuthenticateLoginCallbackHandler

listener.name.sasl_plaintext.oauthbearer.sasl.server.callback.handler.class=com.oauth2.security.oauthbearer.OAuthAuthenticateValidatorCallbackHandler

########## SECURITY using OAUTHBEARER authentication ###############

 

 

--------------------

# vi kafka_server_jaas.conf

KafkaServer {

org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required

LoginStringClaim_sub="<brokerapp-clientId>";

};
-----------------------

## kafka 서버 실행
export KAFKA_OPTS="-Djava.security.auth.login.config=<kafka-binary-dir>/config/kafka_server_jaas.conf -DOAUTH_WITH_SSL=true -DOAUTH_LOGIN_SERVER=<OAuth-server-url> -DOAUTH_LOGIN_ENDPOINT=/oauth2/default/v1/token -DOAUTH_LOGIN_GRANT_TYPE=client_credentials -DOAUTH_LOGIN_SCOPE=kafka -DOAUTH_INTROSPECT_SERVER=<OAuth-server-url> -DOAUTH_INTROSPECT_ENDPOINT=/oauth2/default/v1/introspect -DOAUTH_AUTHORIZATION=Basic%20<encoded-clientId:clientsecret> -DOAUTH_INTROSPECT_AUTHORIZATION=Basic%20<encoded-clientId:clientsecret>"
./bin/kafka-server-start.sh ./config/server.properties