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.
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.
Comments