Desbloqueando a Segurança do Kafka com Listas de Controle de Acesso (ACLs)

O Apache Kafka utiliza Listas de Controle de Acesso (ACLs) como um componente vital de sua infraestrutura de segurança. As ACLs regulam o acesso aos recursos do Kafka, especificando quais usuários ou aplicativos podem executar ações específicas. Ao implementar ACLs, os administradores do Kafka podem impor políticas granulares de controle de acesso, garantindo a integridade e a confidencialidade dos dados dentro de seus clusters Kafka.

As ACLs funcionam como um mecanismo de defesa robusto contra acesso não autorizado, protegendo dados sensíveis e prevenindo atividades maliciosas dentro dos clusters Kafka. Com as ACLs, os administradores podem restringir tópicos, partições, grupos de consumidores e operações administrativas, capacitando-os a aplicar padrões de conformidade e proteger contra violações de dados.

Photo by Patrick Robert Doyle on Unsplash

Kafka com Zookeeper vs. Clusters Kafka baseados em KRaft

A configuração de ACLs no Kafka difere com base na arquitetura do cluster subjacente. Em clusters Kafka tradicionais com Zookeeper, as ACLs são gerenciadas por meio dos nós do Zookeeper, onde as configurações de controle de acesso são armazenadas usando a seguinte classe:

authorizer.class.name=kafka.security.authorizer.AclAuthorizer

No entanto, em clusters Kafka baseados em KRaft, as ACLs são integradas diretamente na configuração do broker, simplificando o gerenciamento e garantindo consistência em todo o cluster. Clusters baseados em KRaft usam uma classe diferente:

authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer

Seguindo o meu artigo anterior sobre a execução do Kafka no Kubernetes com o modo KRaft e autenticação SASL, vou demonstrar como configurar o broker para usar ACLs, como solucionar problemas relacionados às ACLs e apresentar alguns exemplos de ACLs.

Configurando o Kafka para ACLs

A seguir estão os parâmetros que precisamos adicionar ao nosso Broker Kafka anterior para habilitar as ACLs e uma nova configuração de um cliente com permissões de administrador.

Broker

• Habilitar o mecanismo SASL/PLAIN para o CONTROLLER

- name: KAFKA_LISTENER_NAME_CONTROLLER_PLAIN_SASL_JAAS_CONFIG
  value: org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin-secret" user_admin="admin-secret" user_kafkaclient1="kafkaclient1-secret"; 
- name: KAFKA_SASL_MECHANISM_CONTROLLER_PROTOCOL
  value: PLAIN
- name: KAFKA_CONTROLLER_ENABLED_MECHANISMS
  value: PLAIN 

• Atualizar o security.protocol.map do CONTROLLER para usar SASL_PLAINTEXT

- name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
  value: "CONTROLLER:SASL_PLAINTEXT,SASL:SASL_PLAINTEXT"

• Configure o authorizer e super.user

- name: KAFKA_SUPER_USERS
  value: User:admin
- name: KAFKA_AUTHORIZER_CLASS_NAME
  value: org.apache.kafka.metadata.authorizer.StandardAuthorizer
- name: KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND
  value: "false"

Cliente Administrador

• Crie um ConfigMap baseado no arquivo sasl_client.properties:

kubectl create configmap kafka-admin --from-file sasl_admin.properties -n kafka    
kubectl describe configmaps -n kafka kafka-admin

Output:

configmap/kafka-admin created
Name:         kafka-admin
Namespace:    kafka
Labels:       <none>
Annotations:  <none>
    
Data
====
sasl_admin.properties:
----
sasl.mechanism=PLAIN
security.protocol=SASL_PLAINTEXT
    sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
      username="admin" \
      password="admin-secret";
    
BinaryData
====

Events:  <none>

• Monte o ConfigMap como um volume:

    volumeMounts:
        - mountPath: /etc/kafka/secrets/
          name: kafka-admin
    ...
    volumes:
    - name: kafka-admin
      configMap: 
        name: kafka-admin

Criando o deployment

Clone o repo:

git clone https://github.com/rafaelmnatali/kafka-k8s.git
cd acls

Implemente o Kafka usando os seguintes comandos:

kubectl apply -f 00-namespace.yaml
kubectl apply -f 01-kafka.yaml

Verifique a comunicação entre os brokers

Agora deve haver três nós (brokers) Kafka, cada um em execução em pods separados dentro do seu cluster. A resolução de nomes para o headless Service e os três pods dentro do StatefulSet é configurada automaticamente pelo Kubernetes conforme são criados, permitindo a comunicação entre os brokers. Consulte a documentação relacionada para obter mais detalhes sobre esse recurso.

Você pode verificar os logs do primeiro pod com o seguinte comando:

kubectl logs kafka-0

A resoluçāo de nomes para os três pods pode demorar mais tempo do que o pod a iniciar, entāo, você pode ver erros UnknownHostException nos logs durante a inicializaçāo:

WARN [RaftManager nodeId=2] Error connecting to node kafka-1.kafka-headless.kafka.svc.cluster.local:29093 (id: 1 rack: null) (org.apache.kafka.clients.NetworkClient) java.net.UnknownHostException: kafka-1.kafka-headless.kafka.svc.cluster.local         ... 

Eventualmente, cada pod irá resolver os nomes e iniciar com uma mensagem afirmando que o broker foi unfenced:

INFO [Controller 0] Unfenced broker: UnfenceBrokerRecord(id=1, epoch=176) (org.apache.kafka.controller.ClusterControlManager)
Create a topic using the SASL_PLAINTEXT endpoint

Testando Kafka ACLs

Para testar as ACLs, nós vamos fazer o deployment de dois clientes: admin e non-admin. Abra dois terminais e faça o deploy do cliente admin em um e no outro terminal o deploy do cliente non-admin.

Comandos:

kubectl apply -f 02-kafka-client.yaml
kubectl apply -f 03-kafka-admin.yaml

Vamos usar a operação CREATE como exemplo de como solucionar e resolver erros de permissão no Kafka.

Primeiro, crie um tópico a partir do cliente admin:

kubectl exec -it kafka-admin -- bash -c "kafka-topics --bootstrap-server kafka-0.kafka-headless.kafka.svc.cluster.local:9092 -create --topic kafka-admin --command-config /etc/kafka/secrets/sasl_admin.properties"

Saída:

Created topic kafka-admin

Agora, tentamos criar um tópico no cliente non-admin:

kubectl exec -it kafka-client -- bash -c "kafka-topics --bootstrap-server kafka-0.kafka-headless.kafka.svc.cluster.local:9092 -create --topic kafka-client --command-config /etc/kafka/secrets/sasl_client.properties"

Saída:

Error while executing topic command : Authorization failed.

Nosso cliente não tem permissão. Ao examinar os logs do Kafka, é possível ver a seguinte entrada:

kafka-0 kafka [2024-03-13 20:10:17,259] INFO Principal = User:kafkaclient1 is Denied operation = CREATE from host = 10.244.0.9 on resource = Topic:LITERAL:kafka-admin for request = CreateTopics with resourceRefCount = 1 based on rule DefaultDeny (kafka.authorizer.logger)

⚠️ Observe que no registro, esta é uma entrada INFO e não um ERROR.

Crie uma ACL para que o cliente non-admin possa criar tópicos:

kubectl exec -it kafka-admin -- bash -c "kafka-acls --bootstrap-server kafka-0.kafka-headless.kafka.svc.cluster.local:9092 --add --allow-principal User:kafkaclient1 --operation Create --allow-host 10.244.0.9 --cluster --command-config /etc/kafka/secrets/sasl_admin.properties"

Saída:

Adding ACLs for resource `ResourcePattern(resourceType=CLUSTER, name=kafka-cluster, patternType=LITERAL)`: 
        (principal=User:kafkaclient1, host=10.244.0.9, operation=CREATE, permissionType=ALLOW)

Agora, tentamos criar um tópico no cliente non-admin novamente:

kubectl exec -it kafka-client -- bash -c "kafka-topics --bootstrap-server kafka-0.kafka-headless.kafka.svc.cluster.local:9092 -create --topic kafka-client --command-config /etc/kafka/secrets/sasl_client.properties"

Saída:

Created topic kafka-client.

Mais Kafka ACLs exemplos

Read/Write em um tópico:

kafka-acls --bootstrap-server kafka-0.kafka-headless.kafka.svc.cluster.local:9092 --add --allow-principal User:kafkaclient1 --operation Read --operation Write --allow-host 10.244.0.9 --topic kafka-client --command-config /etc/kafka/secrets/sasl_admin.properties

Listar ACLs:

kafka-acls --bootstrap-server kafka-0.kafka-headless.kafka.svc.cluster.local:9092 --list --topic kafka-client --command-config /etc/kafka/secrets/sasl_admin.properties

Remover ACLs:

kafka-acls --bootstrap-server kafka-0.kafka-headless.kafka.svc.cluster.local:9092 --remove --topic kafka-client --command-config /etc/kafka/secrets/sasl_admin.properties

Resumo

As Listas de Controle de Acesso (ACLs) são indispensáveis para garantir a segurança das implantações do Apache Kafka, permitindo que os administradores imponham controles de acesso detalhados e mitiguem os riscos de segurança. Seja gerenciando o Kafka com Zookeeper ou adotando clusters Kafka baseados em KRaft, as ACLs desempenham um papel crucial em garantir a proteção de dados, conformidade e segurança geral do cluster. Ao entender e aproveitar as ACLs de forma eficaz, as organizações podem fortalecer sua infraestrutura Kafka contra acessos não autorizados e ameaças potenciais, reforçando a integridade e confidencialidade de seus pipelines de streaming de eventos.

Referências

• https://rafaelnatali.wixsite.com/rmn-technology/pt/post/executando-o-kafka-no-kubernetes-com-o-modo-kraft-e-autentica%C3%A7%C4%81o-sasl-plain

• https://github.com/rafaelmnatali/kafka-k8s