Renforcement de la sécurité Kafka avec des exemples pratiques

Apache Kafka occupe souvent une place centrale dans les architectures orientées événements, ce qui en fait une cible de choix pour les attaquants. Renforcer Kafka ne consiste pas à activer toutes les fonctions de sécurité simultanément. Il s'agit d'un processus discipliné qui applique les bons contrôles dans le bon ordre, vérifie chaque changement et maintient un chemin de retour propre. Ce guide décrit un flux de travail de renforcement pratique, étape par étape : inventaire, chiffrement TLS, authentification SASL/SCRAM, ACL au moindre privilège, réduction de la surface d'attaque réseau, protection du plan de métadonnées et basculement sécurisé. Chaque exemple est conçu pour être clair, auditable et adaptable à votre environnement.

Inventaire de la version et de l'environnement

Avant de modifier toute configuration, capturez un inventaire concis. Cette base de référence détermine quels contrôles vous activez et dans quel ordre, évitant ainsi les surprises les plus courantes : incompatibilités de protocole, échecs de vérification du nom d'hôte et refus d'ACL pour les composants internes.

• Version et mode Kafka : Notez si vous exécutez Kafka 3.x en mode ZooKeeper ou en mode KRaft (quorum de contrôleurs). KRaft consolide les métadonnées et modifie la configuration des écouteurs du contrôleur.
• Exécution Java et système d'exploitation : Enregistrez le fournisseur JDK et le niveau de correctif (JDK 11+ recommandé), la distribution Linux et la version du noyau.
• Topologie du cluster : Documentez le nombre de courtiers, les noms d'hôte, les adresses IP, les répertoires de données et les paramètres de connaissance des racks.
• Écouteurs et ports actuels : Listez les écouteurs PLAINTEXT, SSL et SASL_SSL existants avec leurs ports.
• Inventaire des clients : Identifiez les applications, les langages et les versions des bibliothèques clientes (Java, Python, Go, etc.). Associez les sujets et les groupes de consommateurs utilisés par chaque client.
• Autorité de certification et racines de confiance : Précisez l'AC interne ou publique, le format du keystore/truststore (PKCS12 ou JKS), les longueurs de clés et les algorithmes.
• Réseau : Notez les pare-feu, les équilibreurs de charge, la configuration DNS, les points de terminaison TLS, les segments internes versus externes, les règles NAT et les politiques de sortie.

Capturez cet inventaire une seule fois et conservez-le accessible pendant le déploiement.

Chemin de configuration sûr : renforcement incrémental

Renforcez par petites étapes vérifiables. Conservez un chemin de configuration fonctionnel et connu comme bon pendant chaque étape et ne le supprimez qu'après vérification.

Étape 0 : Choisir un pilote étroit

Sélectionnez un courtier et un client de test (un producteur et un consommateur). Créez un sujet canari tel que sec.pilot.events. Ce pilote doit être mesurable et facile à inspecter localement avant tout déploiement à l'échelle du cluster.

Étape 1 : Ajouter un écouteur TLS (conserver PLAINTEXT actif)

Objectif : Activer le trafic chiffré sans casser les clients existants.

1. Obtenir les certificats Utilisez votre AC interne. Assurez-vous que les certificats incluent le nom DNS du courtier dans l'extension Subject Alternative Name (SAN). Stockez les fichiers sous /etc/kafka/ssl/ avec des permissions 600 et le propriétaire kafka.

2. Configurer le courtier Ajoutez un nouvel écouteur SSL tout en conservant l'écouteur PLAINTEXT. Mettez à jour server.properties :

# Ajouter de nouveaux écouteurs
listeners=PLAINTEXT://0.0.0.0:9092,SSL://0.0.0.0:9093
advertised.listeners=PLAINTEXT://broker1.example.com:9092,SSL://broker1.example.com:9093
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL
inter.broker.listener.name=SSL

# Paramètres TLS
ssl.keystore.location=/etc/kafka/ssl/broker1.keystore.p12
ssl.keystore.password=changeMeKeystore
ssl.keystore.type=PKCS12
ssl.truststore.location=/etc/kafka/ssl/broker1.truststore.p12
ssl.truststore.password=changeMeTrust
ssl.truststore.type=PKCS12
ssl.client.auth=required

# L'autoriseur sera activé plus tard
allow.everyone.if.no.acl.found=true

3. Redémarrer et vérifier Redémarrez le courtier. Confirmez qu'il démarre sans erreur et que l'écouteur PLAINTEXT reste disponible. Vérifiez la négociation TLS depuis un hôte d'administration qui fait confiance à votre AC :

openssl s_client -connect broker1.example.com:9093 -servername broker1.example.com -tls1_2 -brief

Résultat attendu : Verification: OK et une version TLS et un chiffrement négociés. Si vous voyez verify error: num=20: unable to get local issuer certificate, votre truststore ou votre chaîne de certificats est incorrect.

Étape 2 : Exiger l'authentification client avec SASL/SCRAM (nouveau port)

Objectif : Ajouter un accès authentifié sur un écouteur séparé pour pouvoir tester sans affecter le chemin TLS uniquement.

1. Ajouter un écouteur SASL_SSL Mettez à jour server.properties :

# Ajouter un écouteur SASL_SSL
listeners=PLAINTEXT://0.0.0.0:9092,SSL://0.0.0.0:9093,SASL_SSL://0.0.0.0:9094
advertised.listeners=PLAINTEXT://broker1.example.com:9092,SSL://broker1.example.com:9093,SASL_SSL://broker1.example.com:9094
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_SSL:SASL_SSL
inter.broker.listener.name=SSL

# Activer SASL SCRAM
sasl.enabled.mechanisms=SCRAM-SHA-256,SCRAM-SHA-512
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512

# Préparer l'activation des ACL plus tard
authorizer.class.name=org.apache.kafka.server.authorizer.StandardAuthorizer
allow.everyone.if.no.acl.found=true
super.users=User:admin

2. Fournir la connexion JAAS pour le courtier Créez /etc/kafka/kafka_server_jaas.conf :

KafkaServer {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="broker"
    password="changeMeBroker!";
};

Définissez la variable d'environnement pour le service du courtier :

export KAFKA_OPTS="-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf"

3. Créer des comptes de service avec des identifiants SCRAM Depuis une machine d'administration avec accès au port 9094 :

# Rédacteur
kafka-configs.sh --bootstrap-server broker1.example.com:9094 \
  --alter --add-config 'SCRAM-SHA-512=[password=changeMeNow!]' \
  --entity-type users --entity-name app_writer

# Lecteur
kafka-configs.sh --bootstrap-server broker1.example.com:9094 \
  --alter --add-config 'SCRAM-SHA-512=[password=changeMeRead!]' \
  --entity-type users --entity-name app_reader

4. Tester l'authentification depuis un client Créez client-writer.properties :

security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
ssl.truststore.location=/etc/kafka/ssl/ca.truststore.p12
ssl.truststore.password=changeMeTrust
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="app_writer" password="changeMeNow!";

Produisez vers le sujet canari :

kafka-topics.sh --bootstrap-server broker1.example.com:9094 --create --topic sec.pilot.events --partitions 1 --replication-factor 1
echo 'hello' | kafka-console-producer.sh --bootstrap-server broker1.example.com:9094 \
  --producer.config client-writer.properties --topic sec.pilot.events

Résultat attendu : message accepté. Si vous voyez SASL authentication failed, vérifiez le nom d'utilisateur/mot de passe et confirmez que le client fait confiance au certificat du courtier.

Étape 3 : Appliquer le moindre privilège avec les ACL Kafka

Objectif : Refuser par défaut et n'autoriser que ce dont un principal a besoin.

1. Passer au refus par défaut Dans server.properties :

allow.everyone.if.no.acl.found=false

Redémarrez le courtier.

2. Accorder des droits minimaux aux comptes de service

# Permettre au rédacteur de créer et d'écrire dans le sujet
kafka-acls.sh --bootstrap-server broker1.example.com:9094 \
  --add --allow-principal User:app_writer --operation Create --topic sec.pilot.events
kafka-acls.sh --bootstrap-server broker1.example.com:9094 \
  --add --allow-principal User:app_writer --operation Write --topic sec.pilot.events
kafka-acls.sh --bootstrap-server broker1.example.com:9094 \
  --add --allow-principal User:app_writer --operation IdempotentWrite --cluster

# Permettre au lecteur de lire le sujet et de rejoindre un groupe de consommateurs
kafka-acls.sh --bootstrap-server broker1.example.com:9094 \
  --add --allow-principal User:app_reader --operation Read --topic sec.pilot.events
kafka-acls.sh --bootstrap-server broker1.example.com:9094 \
  --add --allow-principal User:app_reader --operation Read --group sec.pilot.readers

3. Vérifier les ACL Lister les ACL pour le sujet :

kafka-acls.sh --bootstrap-server broker1.example.com:9094 --list --topic sec.pilot.events

Tentez une action non autorisée : essayez de consommer avec app_writer et attendez-vous à un refus dans le client et une entrée de journal ResourceAuthorizationException sur le courtier.

Étape 4 : Réduire l'exposition réseau

• Liez les écouteurs uniquement aux interfaces requises. Préférez lier explicitement aux IP internes, ou utilisez 0.0.0.0 avec des restrictions de pare-feu.
• N'exposez pas PLAINTEXT en dehors d'un réseau d'administration contrôlé.
• Restreignez l'accès aux écouteurs du contrôleur (KRaft) et à ZooKeeper (si utilisé) aux courtiers et administrateurs uniquement.

Référence rapide pour les écouteurs et ports typiques :

Étape 5 : Sécuriser la couche de métadonnées

Mode ZooKeeper :

• Définissez zookeeper.set.acl=true dans server.properties.
• Activez le client TLS ZooKeeper : zookeeper.client.secure=true, zookeeper.ssl.client.enable=true et fournissez keystore/truststore.
• Restreignez ZooKeeper aux sous-réseaux courtier/admin.

Mode KRaft :

• Définissez un écouteur de contrôleur dédié avec TLS, par exemple, listener.security.protocol.map inclut CONTROLLER:SSL et controller.listener.names=CONTROLLER.
• Restreignez le port du contrôleur aux courtiers et contrôleurs uniquement.

Étape 6 : Planifier le basculement et déprécier PLAINTEXT

• Migrez un sous-ensemble de vrais clients vers SASL_SSL avec ACL.
• Observez pendant 24 à 72 heures.
• Si stable, supprimez PLAINTEXT de listeners et advertised.listeners et redémarrez les courtiers pendant une fenêtre de maintenance.

Vérification et diagnostics

Rendez chaque contrôle observable avec des tests concrets.

Vérification TLS

• Négociation et chaîne :

  openssl s_client -connect broker1.example.com:9093 -servername broker1.example.com -tls1_2 -brief

Attendu : Verification: OK et un chiffrement moderne. Si une incompatibilité de nom d'hôte apparaît, corrigez les SAN du certificat.

• Connexion client : Confirmez que les connexions réussissent avec security.protocol=SSL et échouent lors de l'utilisation de PLAINTEXT sur un port restreint.

Vérification SASL/SCRAM

• Succès de l'authentification :

  kafka-console-producer.sh --bootstrap-server broker1.example.com:9094 \
    --producer.config client-writer.properties --topic sec.pilot.events

Tapez quelques lignes et appuyez sur Ctrl-D. Attendu : les journaux du courtier contiennent une entrée similaire à Authenticated principal=User:app_writer via SASL mechanism SCRAM-SHA-512 on listener SASL_SSL.

• Échec de l'authentification : Changez le mot de passe dans client-writer.properties et confirmez que le client signale SASL authentication failed et que le courtier journalise un échec d'authentification avec votre IP source.

Application des ACL

• Test positif : app_reader consomme avec succès uniquement depuis les sujets et groupes autorisés.

  kafka-console-consumer.sh --bootstrap-server broker1.example.com:9094 \
    --consumer.config client-reader.properties --topic sec.pilot.events \
    --group sec.pilot.readers --from-beginning --timeout-ms 5000

Attendu : les messages apparaissent, puis le consommateur quitte sur timeout.

• Test négatif : En utilisant app_reader, essayez de produire vers le sujet et attendez-vous à une erreur d'autorisation.

Cohérence de la configuration

• Vérifiez inter.broker.listener.name et l'alignement des écouteurs sur chaque nœud. Tous les courtiers doivent s'accorder sur l'écouteur inter-courtiers et prendre en charge son protocole.
• Confirmez que les points de terminaison du contrôleur ou de ZooKeeper ne sont accessibles que depuis les courtiers.

Hygiène des journaux et métriques

• Assurez-vous que les journaux incluent le principal, le client-id et l'écouteur dans les entrées d'authentification et d'autorisation.
• Exportez les métriques de requêtes du courtier pour les nouveaux écouteurs afin de repérer les pics d'échecs d'authentification.

Modes de défaillance et récupération

Savoir comment les choses cassent permet de les réparer rapidement.

Échecs TLS

• Symptôme : PKIX path building failed ou certificate_unknown.
• Cause : AC manquant dans le truststore client ou chaîne incomplète.
• Correctif : Importez l'AC émettrice et les intermédiaires dans le truststore client.
• Symptôme : hostname verification failed.
• Cause : Le SAN du certificat ne correspond pas au DNS.
• Correctif : Réémettez les certificats avec les bons SAN DNS. Vérifiez avec :

  openssl x509 -in broker1.crt -noout -text | grep -A1 'Subject Alternative Name'

• Symptôme : Le courtier ne démarre pas après l'activation de SSL.
• Cause : Mauvais mot de passe keystore ou permissions de fichier.
• Correctif : Corrigez les mots de passe, définissez chmod 600 sur keystore/truststore, assurez-vous que le propriétaire est kafka.

Échecs SASL/SCRAM

• Symptôme : Échec d'authentification pour tous les clients.
• Cause : JAAS non chargé ou incompatibilité de mécanisme.
• Correctif : Définissez KAFKA_OPTS avec le chemin JAAS ; confirmez que sasl.enabled.mechanisms inclut votre mécanisme.
• Symptôme : Un seul utilisateur échoue.
• Cause : Identifiant SCRAM manquant ou incorrect.
• Correctif : Réappliquez l'identifiant :

  kafka-configs.sh --bootstrap-server broker1.example.com:9094 \
    --alter --add-config 'SCRAM-SHA-512=[password=newSecret]' \
    --entity-type users --entity-name affected_user

Refus d'ACL

• Symptôme : Les producteurs ou consommateurs s'arrêtent soudainement avec des erreurs d'autorisation.
• Cause : Refus par défaut sans ACL d'autorisation correspondantes.
• Correctif : Ajoutez des ACL minimales pour le sujet et le groupe. Vérifiez avec kafka-acls.sh --list.

Dérive du protocole inter-courtiers

• Symptôme : La réplication s'arrête après le changement d'écouteurs.
• Cause : inter.broker.listener.name défini sur un écouteur que tous les courtiers n'exposent pas.
• Correctif : Assurez-vous que tous les courtiers ont le même écouteur inter-courtiers et protocole de sécurité, puis redémarrage progressif.

Exposition du plan de métadonnées

• Symptôme : Connexions inattendues aux ports ZooKeeper ou du contrôleur KRaft.
• Cause : Lacunes dans le pare-feu.
• Correctif : Restreignez ces ports aux segments courtier et admin uniquement.

Retour arrière et récupération

Gardez le retour arrière simple pendant le pilote.

• Préservez un écouteur PLAINTEXT fonctionnel jusqu'après avoir vérifié SASL_SSL et les ACL. Si SASL bloque les clients, redirigez-les vers PLAINTEXT comme solution de secours à court terme pendant que vous corrigez les identifiants.
• Conservez des sauvegardes de server.properties, des fichiers JAAS et des keystores avant chaque changement.
• Revenez sur les ACL si nécessaire :

  kafka-acls.sh --bootstrap-server broker1.example.com:9094 \
    --remove --allow-principal User:app_writer --operation Write --topic sec.pilot.events

• Supprimez un identifiant utilisateur s'il a été créé par erreur :

  kafka-configs.sh --bootstrap-server broker1.example.com:9094 \
    --alter --delete-config 'SCRAM-SHA-512' --entity-type users --entity-name app_writer

• Solution de dernier recours : restaurez les fichiers de configuration précédents et redémarrez les courtiers dans un ordre contrôlé.

Liste de contrôle opérationnelle

Utilisez cette liste de contrôle compacte pendant le déploiement et pour les revues périodiques.

Avant d'activer de nouveaux contrôles

• [ ] Inventaire de la version Kafka, du mode (ZooKeeper vs KRaft), Java, OS.
• [ ] Confirmer le DNS et les SAN des certificats pour tous les courtiers.
• [ ] Préparer keystore/truststore avec la bonne chaîne d'AC.
• [ ] Créer le sujet pilote et sélectionner les clients pilotes.

TLS

• [ ] Ajouter un écouteur SSL sur un nouveau port et vérifier avec openssl s_client.
• [ ] Définir ssl.client.auth=required et confirmer le TLS mutuel si utilisé.
• [ ] Lier ou pare-feu les écouteurs aux réseaux internes.

SASL/SCRAM

• [ ] Activer sasl.enabled.mechanisms et définir le JAAS du courtier si requis.
• [ ] Créer des utilisateurs par service avec des mots de passe uniques et rotatifs.
• [ ] Tester producteur et consommateur avec security.protocol=SASL_SSL.

ACL

• [ ] Passer à allow.everyone.if.no.acl.found=false seulement après l'existence d'ACL de test.
• [ ] Accorder des permissions minimales de sujet, groupe et cluster.
• [ ] Vérifier les actions autorisées et refusées.

Couche de métadonnées

• [ ] Sécuriser les écouteurs ZooKeeper ou contrôleur avec TLS et ACL réseau.
• [ ] Définir zookeeper.set.acl=true (mode ZooKeeper).

Déclasser PLAINTEXT

• [ ] Migrer les clients progressivement ; surveiller les échecs d'authentification.
• [ ] Supprimer les écouteurs PLAINTEXT après une fenêtre stable.

Secrets et fichiers

• [ ] Permissions keystore/truststore et JAAS 600 ; propriétaire kafka.
• [ ] Aucun secret dans l'historique shell ou fichiers lisibles par tous.

Surveillance et journaux

• [ ] Alerter sur les échecs d'authentification, erreurs TLS et refus d'ACL.
• [ ] Lister périodiquement les ACL et utilisateurs ; supprimer les inutilisés.

Carte de contrôle pratique

Ce tableau relie les objectifs de sécurité courants aux contrôles Kafka spécifiques.

Conclusion

Commencez par un petit pilote facile à inspecter et à annuler. Ajoutez un écouteur TLS, introduisez SASL/SCRAM sur un nouveau port, et activez les ACL en mode refus par défaut seulement après que les comptes et règles sont en place. Vérifiez chaque étape avec des tests concrets : négociations TLS, principaux authentifiés et résultats d'ACL. En vous étendant au cluster complet, supprimez PLAINTEXT et renforcez l'accès réseau aux ports courtier, contrôleur et ZooKeeper. Étendez les mêmes principes aux systèmes qui s'intègrent à Kafka : les processeurs NiFi doivent utiliser SASL_SSL et des principaux spécifiques au service ; les tâches Spark streaming et Structured Streaming doivent s'authentifier et s'exécuter avec des ACL au moindre privilège ; les sources et puits HDFS doivent faire confiance à la même AC et isoler les principaux ; les tâches Airflow qui publient ou consomment doivent utiliser des identifiants par tâche ou par connexion avec des ACL étroites. Avec un chemin mesuré, vérifiable et un plan de retour arrière propre, vous pouvez renforcer Kafka sans interruption et avec des preuves claires que chaque contrôle fonctionne comme prévu.