First commit

emitter/banevents/management/commands/create_join_token.py

@@ -0,0 +1,40 @@
+"""Émet un jeton d'auto-inscription à usage unique pour un nouveau noeud
+(flux "kubeadm join", voir ROADMAP.md et banevents/views.py::join_node).
+À lancer sur le MASTER uniquement — le jeton n'a de sens que contre
+l'endpoint /api/join/ de cette même instance.
+
+Usage :
+    manage.py create_join_token <nom_du_noeud> [--ttl-minutes 60]
+
+Le jeton n'est affiché QU'UNE SEULE FOIS ici, en clair — seul son hash est
+persisté (JoinToken.token_hash). Impossible de le retrouver après coup :
+en cas de perte, en émettre un nouveau (le précédent reste valide jusqu'à
+son expiration ou son utilisation, aucun conflit).
+"""
+from typing import Any
+
+from django.core.management.base import BaseCommand
+
+from banevents.enrollment import DEFAULT_TTL_MINUTES, issue_join_token, join_command
+
+
+class Command(BaseCommand):
+    help = "Émet un jeton d'auto-inscription à usage unique pour un nouveau noeud."
+
+    def add_arguments(self, parser: Any) -> None:
+        parser.add_argument('node_name')
+        parser.add_argument('--ttl-minutes', type=int, default=DEFAULT_TTL_MINUTES)
+
+    def handle(self, *args: Any, **options: Any) -> None:
+        node_name = options['node_name']
+        ttl_minutes = options['ttl_minutes']
+
+        token, _ = issue_join_token(node_name, ttl_minutes)
+
+        self.stdout.write(self.style.SUCCESS(
+            f"Jeton émis pour '{node_name}' (expire dans {ttl_minutes} min, usage unique) :"
+        ))
+        self.stdout.write('')
+        self.stdout.write(f'  {join_command(node_name, token)}')
+        self.stdout.write('')
+        self.stdout.write('À lancer sur le nouveau noeud, puis "sudo make install" comme d\'habitude.')

emitter/banevents/management/commands/master_client.py

@@ -0,0 +1,439 @@
+"""Client MQTT vers le broker master (Phase 2.b/4).
+
+Relaie chaque événement fail2ban publié sur le broker local (même topic
+que banevents.mqtt_listen, fail2ban/+/jail) vers le broker master en mTLS,
+sous fail2ban/<MQTT_MASTER_NODE_NAME>/ban. S'abonne en retour aux commandes
+du master (fail2ban/<MQTT_MASTER_NODE_NAME>/action et
+fail2ban/broadcast/action) et aux mises à jour du roster (Phase 3, voir
+banevents.management.commands.master_listen).
+
+Commandes reconnues (command_handlers ci-dessous) : SYNC_BAN, BAN_ALLPORTS,
+WHITELIST, NOTIFY_ONLY, RATE_LIMIT, ESCALATE. Pour en ajouter une : écrire
+une méthode `execute_<nom>(self, command: dict) -> None` qui termine
+toujours par un appel à `self.report_outcome(cmd, ip, status,
+detail='')`, puis l'ajouter au dict `self.command_handlers` dans
+`handle()`. Toute commande reçue sans handler enregistré est journalisée
+sans action.
+
+Trois formes possibles pour une commande Phase 4 — choisir AVANT
+d'écrire du code (voir ROADMAP.md, Phase 4, "procédure d'ajout") :
+1. Réaction automatique décidée par le master (comme SYNC_BAN/ESCALATE) →
+   un handler ici + une règle `correlation_rule_<nom>` dans
+   master_listen.py.
+2. Décision manuelle diffusée à tous les noeuds (comme
+   WHITELIST/BAN_ALLPORTS/RATE_LIMIT) → un handler ici + ajout aux
+   MANUAL_COMMANDS de publish_command.py.
+3. Action master-only, jamais exécutée par un noeud (comme REPORT_ABUSE)
+   → PAS de handler ici du tout, branchement direct dans
+   publish_command.py.
+
+Ce relais est un process indépendant de mqtt_listen : les deux s'abonnent
+séparément au même topic local, chacun avec sa propre connexion/identité.
+"""
+import ipaddress
+import json
+import os
+import ssl
+import subprocess
+import threading
+import uuid
+from typing import Any
+
+import paho.mqtt.client as mqtt
+import redis
+from asgiref.sync import async_to_sync
+from channels.layers import get_channel_layer
+from django.conf import settings
+from django.core.management.base import BaseCommand, CommandError
+
+from banevents.consumers import GROUP_NAME
+from banevents.models import NodeRegistry
+from banevents.stats import ROSTER_REDIS_DB, ROSTER_REDIS_KEY, ROSTER_TTL_SECONDS
+
+MASTER_SYNC_JAIL = 'master-sync'
+MASTER_RATELIMIT_JAIL = 'master-ratelimit'
+MASTER_ESCALATE_JAIL = 'master-escalate'
+# Intervalle de heartbeat (Phase 3, supervision) : assez court pour qu'une
+# coupure soit détectée en quelques minutes côté dashboard (seuil "hors
+# ligne" = 3x cet intervalle, cf. views.py), assez long pour rester
+# négligeable en trafic MQTT.
+HEARTBEAT_INTERVAL_SECONDS = 60
+
+
+class Command(BaseCommand):
+    help = 'Relaie les événements fail2ban locaux vers le broker master et applique ses commandes (SYNC_BAN).'
+
+    def handle(self, *args: Any, **options: Any) -> None:
+        if not settings.MQTT_MASTER_NODE_NAME:
+            raise CommandError(
+                'MQTT_MASTER_NODE_NAME est vide. Doit être identique au nom '
+                'passé à scripts/generate-node-cert.sh (le CN du certificat '
+                "client) : c'est ce que le master utilisera pour restreindre "
+                'ce noeud via ACL (fail2ban/<nom>/...).'
+            )
+        self.node_topic_prefix = f'fail2ban/{settings.MQTT_MASTER_NODE_NAME}'
+        # Identité fail2ban locale (uuid.getnode(), posée par
+        # f2b_mqtt_action_banisher.py dans chaque payload) — distincte de
+        # MQTT_MASTER_NODE_NAME (l'alias/CN mTLS) : sert uniquement à
+        # reconnaître "ce SYNC_BAN vient de ce noeud", pas à s'authentifier.
+        self.local_node_id = str(uuid.getnode())
+        # Communiqué au master via le heartbeat (voir send_heartbeat) pour
+        # alimenter automatiquement NodeRegistry.dashboard_url côté master
+        # (master_listen.py::handle_heartbeat) — os.environ direct, pas un
+        # Django setting : DASHBOARD_DOMAIN est propre à CE noeud (vide sur
+        # un noeud sans tableau de bord public exposé), même lecture directe
+        # que enrollment.py::join_command.
+        dashboard_domain = os.environ.get('DASHBOARD_DOMAIN', '')
+        self.dashboard_url = f'https://{dashboard_domain}' if dashboard_domain else ''
+        self.heartbeat_timer: threading.Timer | None = None
+        self.command_handlers = {
+            'SYNC_BAN': self.execute_sync_ban,
+            'BAN_ALLPORTS': self.execute_ban_allports,
+            'WHITELIST': self.execute_whitelist,
+            'NOTIFY_ONLY': self.execute_notify_only,
+            'RATE_LIMIT': self.execute_rate_limit,
+            'ESCALATE': self.execute_escalate,
+        }
+        # Cache du roster (Phase 3, cf. master_listen.py) — DB dédiée
+        # (db=2) : db=0 déjà pris par Channels (CHANNEL_LAYERS), db=1 par
+        # Celery (CELERY_BROKER_URL).
+        self.redis_client = redis.Redis(host=settings.REDIS_HOST, port=settings.REDIS_PORT, db=ROSTER_REDIS_DB)
+
+        self.local_client = mqtt.Client(
+            mqtt.CallbackAPIVersion.VERSION2, client_id=f'master-relay-local-{uuid.uuid4().hex[:8]}'
+        )
+        self.local_client.username_pw_set(settings.MQTT_BROKER_USERNAME, settings.MQTT_BROKER_PASSWORD)
+        self.local_client.on_connect = self.on_local_connect
+        self.local_client.on_message = self.on_local_message
+
+        self.master_client = mqtt.Client(
+            mqtt.CallbackAPIVersion.VERSION2, client_id=f'master-relay-master-{uuid.uuid4().hex[:8]}'
+        )
+        # ca_certs=None : le magasin de CA système (défaut Python) sait déjà
+        # vérifier le certificat serveur Let's Encrypt du master. Notre CA
+        # privée (scripts/master-ca-init.sh) ne sert qu'à ce que LE MASTER
+        # vérifie NOTRE certificat client (mTLS) — elle n'a rien à voir avec
+        # la vérification du certificat serveur par CE client, et la fournir
+        # ici écrase le magasin par défaut au lieu de s'y ajouter, faisant
+        # échouer la validation du certificat serveur (constaté :
+        # SSLCertVerificationError "unable to get local issuer certificate").
+        self.master_client.tls_set(
+            certfile=settings.MQTT_MASTER_CLIENT_CERT,
+            keyfile=settings.MQTT_MASTER_CLIENT_KEY,
+            tls_version=ssl.PROTOCOL_TLSv1_2,
+        )
+        self.master_client.on_connect = self.on_master_connect
+        self.master_client.on_message = self.on_master_message
+        # Callback dédié (pas on_master_message) : le roster est un flux
+        # séparé des commandes (action), pas la peine de faire cohabiter
+        # deux formats de message dans un seul handler générique.
+        self.master_client.message_callback_add('fail2ban/broadcast/roster', self.on_roster_message)
+
+        self.stdout.write(f'Connexion locale à {settings.MQTT_BROKER_HOST}:{settings.MQTT_BROKER_PORT}')
+        self.local_client.connect(settings.MQTT_BROKER_HOST, settings.MQTT_BROKER_PORT, keepalive=60)
+        self.local_client.loop_start()
+
+        # connect_async + retry_first_connection : si le master est injoignable
+        # au démarrage (réseau, certificat pas encore en place, ...),
+        # loop_forever() réessaie tout seul au lieu de faire planter toute la
+        # commande (y compris le relais local, qui fonctionnerait pourtant
+        # très bien sans lui). Sans retry_first_connection, un échec de LA
+        # toute première tentative fait remonter l'exception au lieu d'être
+        # réessayé (constaté : ConnectionRefusedError non rattrapée).
+        self.stdout.write(f'Connexion master à {settings.MQTT_MASTER_HOST}:{settings.MQTT_MASTER_PORT}')
+        self.master_client.connect_async(settings.MQTT_MASTER_HOST, settings.MQTT_MASTER_PORT, keepalive=60)
+        self.master_client.reconnect_delay_set(min_delay=1, max_delay=30)
+        self.master_client.loop_forever(retry_first_connection=True)
+
+    def on_local_connect(
+        self,
+        client: mqtt.Client,
+        userdata: Any,
+        flags: mqtt.ConnectFlags,
+        reason_code: mqtt.ReasonCode,
+        properties: Any = None,
+    ) -> None:
+        self.stdout.write(self.style.SUCCESS(f'Connecté au broker local ({reason_code})'))
+        client.subscribe(settings.MQTT_TOPIC_SUBSCRIBE)
+
+    def on_local_message(self, client: mqtt.Client, userdata: Any, message: mqtt.MQTTMessage) -> None:
+        try:
+            payload = json.loads(message.payload.decode('utf-8'))
+        except (json.JSONDecodeError, UnicodeDecodeError) as e:
+            self.stderr.write(self.style.ERROR(f'Message local invalide : {e}'))
+            return
+
+        # Le topic doit correspondre au CN du certificat mTLS de ce process
+        # (MQTT_MASTER_NODE_NAME), jamais au "node" interne du payload
+        # (uuid.getnode() côté fail2ban, un simple identifiant matériel) : le
+        # master ACL chaque connexion sur son identité de certificat, donc
+        # un topic qui ne correspond pas à ce CN serait rejeté.
+        topic = f'{self.node_topic_prefix}/ban'
+        self.master_client.publish(topic, json.dumps(payload), qos=1)
+        self.stdout.write(f'Relayé vers le master : {topic} {payload.get("action")} {payload.get("ip")}')
+
+    def on_master_connect(
+        self,
+        client: mqtt.Client,
+        userdata: Any,
+        flags: mqtt.ConnectFlags,
+        reason_code: mqtt.ReasonCode,
+        properties: Any = None,
+    ) -> None:
+        self.stdout.write(self.style.SUCCESS(f'Connecté au master ({reason_code})'))
+        client.subscribe(f'{self.node_topic_prefix}/action')
+        client.subscribe('fail2ban/broadcast/action')
+        client.subscribe('fail2ban/broadcast/roster')
+        # on_master_connect peut se redéclencher à chaque reconnexion :
+        # annuler toute chaîne de timer précédente pour ne pas en empiler
+        # plusieurs en parallèle (heartbeats en double).
+        if self.heartbeat_timer is not None:
+            self.heartbeat_timer.cancel()
+        self.send_heartbeat()
+
+    def send_heartbeat(self) -> None:
+        # Le segment de topic (fail2ban/<MQTT_MASTER_NODE_NAME>/heartbeat)
+        # est l'alias/CN mTLS, imposé par l'ACL (pattern write
+        # fail2ban/%u/heartbeat) — jamais le même identifiant que
+        # NodeRegistry.node_id (uuid.getnode() brut, cf. commentaire plus
+        # haut sur ces deux identités distinctes). L'id brut voyage donc
+        # dans le payload, exactement comme /ban le fait déjà.
+        self.master_client.publish(
+            f'{self.node_topic_prefix}/heartbeat',
+            json.dumps({'node': self.local_node_id, 'dashboard_url': self.dashboard_url}),
+            qos=0,
+        )
+        self.heartbeat_timer = threading.Timer(HEARTBEAT_INTERVAL_SECONDS, self.send_heartbeat)
+        self.heartbeat_timer.daemon = True
+        self.heartbeat_timer.start()
+
+    def report_outcome(self, cmd: str, ip: str, status: str, detail: str = '') -> None:
+        """Diffuse l'issue d'une commande reçue du master vers ses deux
+        destinations : accusé de réception MQTT vers le master
+        (fail2ban/<noeud>/ack) et notification WebSocket vers ce dashboard
+        (même canal que les BanEvent, cf. signals.py) — un SYNC_BAN réussi
+        finit par apparaître indirectement via le BanEvent qu'il déclenche,
+        mais un échec ou une commande non gérée ne laissaient jusqu'ici
+        aucune trace visible hors journalctl. Point d'entrée unique pour
+        toute nouvelle commande : chaque execute_<nom> doit terminer par un
+        appel ici plutôt que de publier séparément vers les deux canaux."""
+        payload = {'cmd': cmd, 'ip': ip, 'status': status, 'detail': detail}
+        self.master_client.publish(f'{self.node_topic_prefix}/ack', json.dumps(payload), qos=1)
+        channel_layer = get_channel_layer()
+        if channel_layer is not None:
+            async_to_sync(channel_layer.group_send)(
+                GROUP_NAME, {'type': 'command.notification', 'payload': payload},
+            )
+
+    def on_master_message(self, client: mqtt.Client, userdata: Any, message: mqtt.MQTTMessage) -> None:
+        try:
+            command = json.loads(message.payload.decode('utf-8'))
+        except (json.JSONDecodeError, UnicodeDecodeError) as e:
+            self.stderr.write(self.style.ERROR(f'Commande master invalide : {e}'))
+            return
+
+        cmd = command.get('cmd')
+        handler = self.command_handlers.get(cmd)
+        if handler is not None:
+            handler(command)
+        else:
+            # Phase 4 restante (ESCALATE, RATE_LIMIT, REPORT_ABUSE,
+            # REQUEST_GEOLOCATE) : pas de handler enregistré, journalisée
+            # seulement.
+            self.stdout.write(self.style.WARNING(f'Commande reçue du master (non gérée) : {command}'))
+            self.report_outcome(str(cmd or '?'), '', 'unhandled')
+
+    def _banip_via_jail(self, jail: str, ip: str) -> tuple[bool, str]:
+        """Bannit `ip` via la jail `jail` (`fail2ban-client set <jail>
+        banip <ip>`) — primitive partagée par toute commande qui se
+        résume à "bannir cette IP dans une jail dédiée à injection
+        manuelle" (master-sync/SYNC_BAN, master-sync/BAN_ALLPORTS,
+        master-ratelimit/RATE_LIMIT, master-escalate/ESCALATE) : seuls le
+        nom de la jail et l'origine de la décision changent. Retourne
+        (succès, detail — message d'erreur si échec)."""
+        try:
+            result = subprocess.run(
+                ['sudo', settings.FAIL2BAN_CLIENT_PATH, 'set', jail, 'banip', ip],
+                capture_output=True, text=True, timeout=10,
+            )
+        except (subprocess.SubprocessError, OSError) as e:
+            return False, str(e)
+        if result.returncode == 0:
+            return True, ''
+        return False, result.stderr.strip() or result.stdout.strip()
+
+    def _list_active_jails(self) -> list[str]:
+        """Parse la sortie de `fail2ban-client status` ("Jail list: a, b, c")
+        — pas d'option JSON native côté fail2ban-client pour cette info."""
+        try:
+            result = subprocess.run(
+                ['sudo', settings.FAIL2BAN_CLIENT_PATH, 'status'],
+                capture_output=True, text=True, timeout=10, check=True,
+            )
+        except (subprocess.SubprocessError, OSError, subprocess.CalledProcessError) as e:
+            self.stderr.write(self.style.ERROR(f'Impossible de lister les jails actives : {e}'))
+            return []
+        for line in result.stdout.splitlines():
+            if 'Jail list:' in line:
+                return [j.strip() for j in line.split(':', 1)[1].split(',') if j.strip()]
+        return []
+
+    def execute_sync_ban(self, command: dict[str, Any]) -> None:
+        if str(command.get('origin_node', '')) == self.local_node_id:
+            self.stdout.write(f'SYNC_BAN ignoré (origine = ce noeud) : {command.get("ip")}')
+            return
+
+        try:
+            ip = str(ipaddress.ip_address(command.get('ip', '')))
+        except (ValueError, TypeError):
+            self.stderr.write(self.style.ERROR(f'SYNC_BAN : IP invalide reçue : {command.get("ip")!r}'))
+            self.report_outcome('SYNC_BAN', str(command.get('ip', '')), 'error', 'IP invalide')
+            return
+
+        ok, detail = self._banip_via_jail(MASTER_SYNC_JAIL, ip)
+        if ok:
+            self.stdout.write(self.style.SUCCESS(f'SYNC_BAN appliqué : {ip} (jail {MASTER_SYNC_JAIL})'))
+            self.report_outcome('SYNC_BAN', ip, 'ok')
+        else:
+            self.stderr.write(self.style.ERROR(f'SYNC_BAN échec pour {ip} : {detail}'))
+            self.report_outcome('SYNC_BAN', ip, 'error', detail)
+
+    def execute_ban_allports(self, command: dict[str, Any]) -> None:
+        """Décision ciblée explicite (pas une synchro auto : pas de check
+        origin_node — déclenchée à la main via publish_command.py, doit
+        s'appliquer sur tous les noeuds qui la reçoivent, y compris
+        l'origine si jamais elle en porte une)."""
+        try:
+            ip = str(ipaddress.ip_address(command.get('ip', '')))
+        except (ValueError, TypeError):
+            self.stderr.write(self.style.ERROR(f'BAN_ALLPORTS : IP invalide reçue : {command.get("ip")!r}'))
+            self.report_outcome('BAN_ALLPORTS', str(command.get('ip', '')), 'error', 'IP invalide')
+            return
+
+        ok, detail = self._banip_via_jail(MASTER_SYNC_JAIL, ip)
+        if ok:
+            self.stdout.write(self.style.SUCCESS(f'BAN_ALLPORTS appliqué : {ip} (jail {MASTER_SYNC_JAIL})'))
+            self.report_outcome('BAN_ALLPORTS', ip, 'ok')
+        else:
+            self.stderr.write(self.style.ERROR(f'BAN_ALLPORTS échec pour {ip} : {detail}'))
+            self.report_outcome('BAN_ALLPORTS', ip, 'error', detail)
+
+    def execute_whitelist(self, command: dict[str, Any]) -> None:
+        """Ajoute l'IP à ignoreip sur toutes les jails actives + la
+        débannit si elle l'était. Limitation connue : `addignoreip` est un
+        réglage en mémoire, non persistant — perdu au prochain redémarrage
+        de fail2ban.service (install.sh ne touche pas jail.local donc un
+        `install.sh install` ne l'efface pas, mais un restart manuel du
+        service, oui). Persister ça proprement (fichier dédié non écrasé
+        par install.sh, relu au démarrage) reste à faire séparément."""
+        try:
+            ip = str(ipaddress.ip_address(command.get('ip', '')))
+        except (ValueError, TypeError):
+            self.stderr.write(self.style.ERROR(f'WHITELIST : IP invalide reçue : {command.get("ip")!r}'))
+            self.report_outcome('WHITELIST', str(command.get('ip', '')), 'error', 'IP invalide')
+            return
+
+        jails = self._list_active_jails()
+        if not jails:
+            self.report_outcome('WHITELIST', ip, 'error', 'aucune jail active trouvée')
+            return
+        for jail in jails:
+            subprocess.run(
+                ['sudo', settings.FAIL2BAN_CLIENT_PATH, 'set', jail, 'addignoreip', ip],
+                capture_output=True, text=True, timeout=10,
+            )
+            # Échec ignoré : l'IP peut légitimement ne pas être bannie
+            # dans cette jail précise.
+            subprocess.run(
+                ['sudo', settings.FAIL2BAN_CLIENT_PATH, 'set', jail, 'unbanip', ip],
+                capture_output=True, text=True, timeout=10,
+            )
+        self.stdout.write(self.style.SUCCESS(f'WHITELIST appliqué : {ip} ({len(jails)} jails)'))
+        self.report_outcome('WHITELIST', ip, 'ok')
+
+    def execute_notify_only(self, command: dict[str, Any]) -> None:
+        """Aucune action système — la visibilité "alerte admin" demandée
+        pour cette commande (cf. ROADMAP.md) est déjà couverte par le
+        toast que report_outcome déclenche côté dashboard."""
+        ip = str(command.get('ip', ''))
+        self.stdout.write(f'NOTIFY_ONLY reçu : {ip}')
+        self.report_outcome('NOTIFY_ONLY', ip, 'ok')
+
+    def execute_rate_limit(self, command: dict[str, Any]) -> None:
+        """Décision ciblée explicite (manuelle, via publish_command.py) :
+        throttle au lieu d'un blocage total, jail dédiée master-ratelimit
+        (banaction f2b-iptables-hashlimit)."""
+        try:
+            ip = str(ipaddress.ip_address(command.get('ip', '')))
+        except (ValueError, TypeError):
+            self.stderr.write(self.style.ERROR(f'RATE_LIMIT : IP invalide reçue : {command.get("ip")!r}'))
+            self.report_outcome('RATE_LIMIT', str(command.get('ip', '')), 'error', 'IP invalide')
+            return
+
+        ok, detail = self._banip_via_jail(MASTER_RATELIMIT_JAIL, ip)
+        if ok:
+            self.stdout.write(self.style.SUCCESS(f'RATE_LIMIT appliqué : {ip} (jail {MASTER_RATELIMIT_JAIL})'))
+            self.report_outcome('RATE_LIMIT', ip, 'ok')
+        else:
+            self.stderr.write(self.style.ERROR(f'RATE_LIMIT échec pour {ip} : {detail}'))
+            self.report_outcome('RATE_LIMIT', ip, 'error', detail)
+
+    def execute_escalate(self, command: dict[str, Any]) -> None:
+        """Auto-publiée par master_listen.py (correlation_rule_escalate)
+        quand une IP est bannie indépendamment sur plusieurs noeuds
+        distincts. Contrairement à SYNC_BAN, pas de check origin_node :
+        doit s'appliquer aussi sur le(s) noeud(s) d'origine (leur ban
+        initial n'est pas permanent, master-escalate est une jail
+        séparée, pas de conflit à réappliquer)."""
+        try:
+            ip = str(ipaddress.ip_address(command.get('ip', '')))
+        except (ValueError, TypeError):
+            self.stderr.write(self.style.ERROR(f'ESCALATE : IP invalide reçue : {command.get("ip")!r}'))
+            self.report_outcome('ESCALATE', str(command.get('ip', '')), 'error', 'IP invalide')
+            return
+
+        ok, detail = self._banip_via_jail(MASTER_ESCALATE_JAIL, ip)
+        if ok:
+            self.stdout.write(self.style.SUCCESS(f'ESCALATE appliqué : {ip} (jail {MASTER_ESCALATE_JAIL}, permanent)'))
+            self.report_outcome('ESCALATE', ip, 'ok')
+        else:
+            self.stderr.write(self.style.ERROR(f'ESCALATE échec pour {ip} : {detail}'))
+            self.report_outcome('ESCALATE', ip, 'error', detail)
+
+    def on_roster_message(self, client: mqtt.Client, userdata: Any, message: mqtt.MQTTMessage) -> None:
+        """Roster périodique publié par master_listen.py (Phase 3) : les
+        noeuds connus se propagent dans NodeRegistry (déjà le modèle lu par
+        la sidebar — un client voit ainsi les autres noeuds, pas que
+        lui-même) ; les stats globales jail/pays (pas des lignes de table,
+        des agrégats transitoires) vont en cache Redis, lu par views.py
+        avec repli sur le calcul local si absent/périmé."""
+        try:
+            roster = json.loads(message.payload.decode('utf-8'))
+        except (json.JSONDecodeError, UnicodeDecodeError) as e:
+            self.stderr.write(self.style.ERROR(f'Roster invalide : {e}'))
+            return
+
+        for node in roster.get('nodes', []):
+            node_id = node.get('node_id')
+            if not node_id:
+                continue
+            # filter().update() plutôt que update_or_create() : last_seen a
+            # auto_now=True, qui écraserait la vraie valeur du master par
+            # "maintenant" si on passait par .save() (déclenché en interne
+            # par update_or_create) — .update() fait une requête SQL directe,
+            # sans repasser par la logique auto_now du champ.
+            fields = {
+                'alias': node.get('alias', ''),
+                'country': node.get('country', ''),
+                'dashboard_url': node.get('dashboard_url', ''),
+            }
+            if node.get('last_seen'):
+                fields['last_seen'] = node['last_seen']
+            if not NodeRegistry.objects.filter(node_id=node_id).update(**fields):
+                NodeRegistry.objects.get_or_create(node_id=node_id, defaults=fields)
+
+        self.redis_client.set(
+            ROSTER_REDIS_KEY,
+            json.dumps({'top_jails': roster.get('top_jails', []), 'top_countries': roster.get('top_countries', [])}),
+            ex=ROSTER_TTL_SECONDS,
+        )

emitter/banevents/management/commands/master_listen.py

@@ -0,0 +1,237 @@
+"""Client MQTT côté master (Phase 3/4, minimal) : écoute fail2ban/+/ban
+(tous les noeuds connectés) en mTLS, persiste chaque événement dans le
+même modèle BanEvent que mqtt_listen — dashboard multi-noeuds unifié via
+le filtre par noeud déjà présent, sans rien dupliquer — et applique un
+registre de règles de corrélation (self.correlation_rules, construit
+dans handle()) à chaque BAN reçu : SYNC_BAN (propagation systématique à
+tous les noeuds) et ESCALATE (ban permanent si l'IP est bannie
+indépendamment sur plusieurs noeuds distincts, cf.
+correlation_rule_escalate). Pour ajouter une future règle
+auto-déclenchée : écrire une méthode `correlation_rule_<nom>(self,
+client, payload) -> None` et l'ajouter à self.correlation_rules dans
+handle() — voir ROADMAP.md (Phase 4, "procédure d'ajout") pour les deux
+autres formes possibles (commande manuelle diffusée, action master-only)
+selon la nature de la nouvelle commande.
+
+Publie aussi périodiquement un "roster" (fail2ban/broadcast/roster, Phase
+3) : la liste des noeuds connus (NodeRegistry) + les compteurs jail/pays
+globaux — sans ça, la sidebar d'un noeud client n'aurait jamais que SES
+propres données (aucun autre process que celui-ci ne maintient
+NodeRegistry à jour pour plus d'un noeud). master_client.py de chaque
+noeud s'y abonne et republie localement (NodeRegistry + cache Redis, voir
+banevents/stats.py).
+
+Identité mTLS dédiée (MQTT_MASTER_LISTENER_CERT/KEY, CN "master-internal"
+par convention), distincte de celle de master_client : ce process doit à
+la fois LIRE tous les noeuds (fail2ban/+/ban) et ÉCRIRE sur le canal de
+diffusion — deux droits plus larges que ceux d'un noeud pris
+individuellement, donc son propre certificat (voir master-acl.conf).
+"""
+import datetime
+import json
+import ssl
+import threading
+import uuid
+from typing import Any
+
+import paho.mqtt.client as mqtt
+from django.conf import settings
+from django.core.management.base import BaseCommand
+from django.core.serializers.json import DjangoJSONEncoder
+from django.utils import timezone
+
+from banevents.ingestion import save_ban_event
+from banevents.models import BanEvent, NodeRegistry
+from banevents.stats import top_jails_and_countries
+
+MASTER_BAN_TOPIC = 'fail2ban/+/ban'
+MASTER_HEARTBEAT_TOPIC = 'fail2ban/+/heartbeat'
+MASTER_ACK_TOPIC = 'fail2ban/+/ack'
+BROADCAST_ACTION_TOPIC = 'fail2ban/broadcast/action'
+BROADCAST_ROSTER_TOPIC = 'fail2ban/broadcast/roster'
+# Même ordre de grandeur que HEARTBEAT_INTERVAL_SECONDS (master_client.py) —
+# pas besoin d'être plus réactif, la sidebar n'a rien de temps réel critique.
+ROSTER_INTERVAL_SECONDS = 60
+# ESCALATE (Phase 4) : seuil de corrélation cross-node.
+ESCALATE_NODE_THRESHOLD = 2
+ESCALATE_WINDOW_HOURS = 24
+# Jails "réactives" (déclenchées par une commande, pas par une détection
+# locale indépendante) — exclues du comptage ESCALATE, sinon la cascade
+# SYNC_BAN habituelle (un ban propagé rebondit une fois avant de
+# s'arrêter, cf. ROADMAP.md) se compterait elle-même comme une 2e
+# détection indépendante.
+REACTIVE_JAIL_NAMES = {'master-sync', 'master-ratelimit', 'master-escalate'}
+
+
+class Command(BaseCommand):
+    help = (
+        "Écoute fail2ban/+/ban sur le broker master, enregistre chaque événement, "
+        "et publie un SYNC_BAN en diffusion générale pour chaque BAN reçu."
+    )
+
+    def handle(self, *args: Any, **options: Any) -> None:
+        self.roster_timer: threading.Timer | None = None
+        self.correlation_rules = [self.correlation_rule_sync_ban, self.correlation_rule_escalate]
+        client = mqtt.Client(
+            mqtt.CallbackAPIVersion.VERSION2, client_id=f'master-listen-{uuid.uuid4().hex[:8]}'
+        )
+        # ca_certs=None : voir master_client.py — le magasin de CA système
+        # sait déjà vérifier le certificat serveur Let's Encrypt du master.
+        client.tls_set(
+            certfile=settings.MQTT_MASTER_LISTENER_CERT,
+            keyfile=settings.MQTT_MASTER_LISTENER_KEY,
+            tls_version=ssl.PROTOCOL_TLSv1_2,
+        )
+        client.on_connect = self.on_connect
+        client.on_message = self.on_message
+
+        self.stdout.write(f'Connexion master à {settings.MQTT_MASTER_HOST}:{settings.MQTT_MASTER_PORT}')
+        client.connect_async(settings.MQTT_MASTER_HOST, settings.MQTT_MASTER_PORT, keepalive=60)
+        client.reconnect_delay_set(min_delay=1, max_delay=30)
+        client.loop_forever(retry_first_connection=True)
+
+    def on_connect(
+        self,
+        client: mqtt.Client,
+        userdata: Any,
+        flags: mqtt.ConnectFlags,
+        reason_code: mqtt.ReasonCode,
+        properties: Any = None,
+    ) -> None:
+        self.stdout.write(self.style.SUCCESS(f'Connecté au master ({reason_code})'))
+        client.subscribe(MASTER_BAN_TOPIC)
+        client.subscribe(MASTER_HEARTBEAT_TOPIC)
+        client.subscribe(MASTER_ACK_TOPIC)
+        # on_connect peut se redéclencher à chaque reconnexion : annuler
+        # tout timer précédent pour ne pas empiler plusieurs chaînes en
+        # parallèle (rosters en double), même précaution que le heartbeat
+        # de master_client.py.
+        if self.roster_timer is not None:
+            self.roster_timer.cancel()
+        self.publish_roster(client)
+
+    def on_message(self, client: mqtt.Client, userdata: Any, message: mqtt.MQTTMessage) -> None:
+        # Topics séparés par suffixe (fail2ban/<node>/ban|heartbeat|ack),
+        # jamais un wildcard multi-niveau fail2ban/+/+ — reste explicite,
+        # cohérent avec le reste de l'ACL (master-acl.conf).
+        topic_kind = message.topic.rsplit('/', 1)[-1]
+        if topic_kind == 'ban':
+            self.handle_ban(client, message)
+        elif topic_kind == 'heartbeat':
+            self.handle_heartbeat(message)
+        elif topic_kind == 'ack':
+            self.handle_ack(message)
+
+    def handle_ban(self, client: mqtt.Client, message: mqtt.MQTTMessage) -> None:
+        try:
+            payload = json.loads(message.payload.decode('utf-8'))
+            # Segment du topic (fail2ban/<CN>/ban) = CN du certificat client
+            # = node_name choisi à l'inscription (sign-node-csr.sh) — sert
+            # uniquement de suggestion d'alias à la création d'un NodeRegistry
+            # inédit (voir save_ban_event) ; payload['node'] (uuid.getnode())
+            # reste le vrai node_id, ces deux identifiants ne coïncident pas.
+            alias_hint = message.topic.split('/')[1]
+            save_ban_event(payload, alias_hint=alias_hint)
+        except Exception as e:
+            self.stderr.write(self.style.ERROR(f'Message master invalide ({message.topic}) : {e}'))
+            return
+        self.stdout.write(
+            f'Événement master enregistré : noeud={payload.get("node")} '
+            f'{payload.get("name")} {payload.get("action")} {payload.get("ip")}'
+        )
+
+        if payload.get('action') == 'BAN':
+            for rule in self.correlation_rules:
+                rule(client, payload)
+
+    def handle_heartbeat(self, message: mqtt.MQTTMessage) -> None:
+        # Le segment de topic (fail2ban/<node>/heartbeat) est le CN du
+        # certificat client (MQTT_MASTER_NODE_NAME), PAS le même
+        # identifiant que NodeRegistry.node_id (uuid.getnode() brut, cf.
+        # payload des BanEvent) — l'id brut voyage donc dans le payload,
+        # comme pour /ban.
+        try:
+            payload = json.loads(message.payload.decode('utf-8'))
+            node_id = str(payload['node'])
+        except (json.JSONDecodeError, UnicodeDecodeError, KeyError) as e:
+            self.stderr.write(self.style.ERROR(f'Heartbeat invalide ({message.topic}) : {e}'))
+            return
+        # .update() plutôt que get_or_create() : un noeud pas encore connu
+        # de NodeRegistry (jamais banni depuis son déploiement) est un
+        # no-op silencieux ici — il apparaîtra dès son premier ban, pas la
+        # peine de le créer juste pour un heartbeat.
+        update_fields: dict[str, Any] = {'last_seen': timezone.now()}
+        # dashboard_url (DASHBOARD_DOMAIN de CE noeud, cf. master_client.py::
+        # send_heartbeat) : resynchronisé à CHAQUE heartbeat, pas seulement à
+        # la création — contrairement à alias/country (des labels choisis
+        # une fois), c'est une donnée technique qui doit refléter le .env
+        # réel du noeud. Absent/vide (noeud sans tableau de bord public
+        # exposé) => clé omise, ne jamais écraser une valeur déjà connue
+        # avec du vide.
+        dashboard_url = payload.get('dashboard_url')
+        if dashboard_url:
+            update_fields['dashboard_url'] = dashboard_url
+        NodeRegistry.objects.filter(node_id=node_id).update(**update_fields)
+
+    def handle_ack(self, message: mqtt.MQTTMessage) -> None:
+        node_id = message.topic.split('/')[1]
+        try:
+            payload = json.loads(message.payload.decode('utf-8'))
+        except (json.JSONDecodeError, UnicodeDecodeError) as e:
+            self.stderr.write(self.style.ERROR(f'ACK invalide ({message.topic}) : {e}'))
+            return
+        self.stdout.write(
+            f'ACK reçu de {node_id} : {payload.get("cmd")} {payload.get("ip")} -> '
+            f'{payload.get("status")} {payload.get("detail", "")}'.rstrip()
+        )
+
+    def publish_roster(self, client: mqtt.Client) -> None:
+        nodes = list(
+            NodeRegistry.objects.all().values('node_id', 'alias', 'country', 'dashboard_url', 'last_seen')
+        )
+        roster = {'nodes': nodes, **top_jails_and_countries()}
+        client.publish(BROADCAST_ROSTER_TOPIC, json.dumps(roster, cls=DjangoJSONEncoder), qos=0)
+        self.roster_timer = threading.Timer(ROSTER_INTERVAL_SECONDS, self.publish_roster, args=[client])
+        self.roster_timer.daemon = True
+        self.roster_timer.start()
+
+    def correlation_rule_sync_ban(self, client: mqtt.Client, payload: dict[str, Any]) -> None:
+        command = {
+            'cmd': 'SYNC_BAN',
+            'ip': payload.get('ip'),
+            'ttl': payload.get('bantime'),
+            'reason': payload.get('name', ''),
+            'origin_node': payload.get('node', ''),
+        }
+        client.publish(BROADCAST_ACTION_TOPIC, json.dumps(command), qos=1)
+        self.stdout.write(f'SYNC_BAN publié : {command}')
+
+    def correlation_rule_escalate(self, client: mqtt.Client, payload: dict[str, Any]) -> None:
+        """Publie ESCALATE (ban permanent, jail master-escalate côté
+        noeuds) si `ip` a été bannie de façon INDÉPENDANTE (hors jails
+        REACTIVE_JAIL_NAMES — sinon la cascade SYNC_BAN habituelle se
+        compterait elle-même) sur au moins ESCALATE_NODE_THRESHOLD noeuds
+        distincts dans les dernières ESCALATE_WINDOW_HOURS heures."""
+        ip = payload.get('ip')
+        if not ip:
+            return
+
+        window_start = timezone.now() - datetime.timedelta(hours=ESCALATE_WINDOW_HOURS)
+        distinct_nodes = (
+            BanEvent.objects.filter(ip_address=ip, action='BAN', received_at__gte=window_start)
+            .exclude(jail_name__in=REACTIVE_JAIL_NAMES)
+            .values('node').distinct().count()
+        )
+        if distinct_nodes < ESCALATE_NODE_THRESHOLD:
+            return
+
+        command = {
+            'cmd': 'ESCALATE',
+            'ip': ip,
+            'reason': payload.get('name', ''),
+            'origin_node': payload.get('node', ''),
+        }
+        client.publish(BROADCAST_ACTION_TOPIC, json.dumps(command), qos=1)
+        self.stdout.write(self.style.WARNING(
+            f'ESCALATE publié ({distinct_nodes} noeuds indépendants en {ESCALATE_WINDOW_HOURS}h) : {command}'
+        ))

emitter/banevents/management/commands/mqtt_listen.py

@@ -0,0 +1,54 @@
+"""Client MQTT local : ingère les événements fail2ban publiés par
+f2b-mqtt-action-banisher et les persiste en base (modèle BanEvent)."""
+import json
+import uuid
+from typing import Any
+
+import paho.mqtt.client as mqtt
+from django.conf import settings
+from django.core.management.base import BaseCommand
+
+from banevents.ingestion import save_ban_event
+
+
+class Command(BaseCommand):
+    help = "Écoute le broker MQTT local et enregistre chaque événement fail2ban reçu."
+
+    def handle(self, *args: Any, **options: Any) -> None:
+        # client_id unique par process : deux instances partageant le même id
+        # MQTT (deux déploiements, ou un test local pointé sur le même
+        # broker) s'éjectent mutuellement en boucle sinon (constaté en
+        # prod : une instance de dev oubliée connectée via le VPN et le
+        # service systemd du VPS se disputaient le même id).
+        client_id = f'emitter-mqtt-listen-{uuid.uuid4().hex[:8]}'
+        client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id)
+        client.username_pw_set(settings.MQTT_BROKER_USERNAME, settings.MQTT_BROKER_PASSWORD)
+        client.on_connect = self.on_connect
+        client.on_message = self.on_message
+
+        self.stdout.write(
+            f'Connexion à {settings.MQTT_BROKER_HOST}:{settings.MQTT_BROKER_PORT} '
+            f'(topic {settings.MQTT_TOPIC_SUBSCRIBE})'
+        )
+        client.connect(settings.MQTT_BROKER_HOST, settings.MQTT_BROKER_PORT, keepalive=60)
+        client.loop_forever()
+
+    def on_connect(
+        self,
+        client: mqtt.Client,
+        userdata: Any,
+        flags: mqtt.ConnectFlags,
+        reason_code: mqtt.ReasonCode,
+        properties: Any = None,
+    ) -> None:
+        self.stdout.write(self.style.SUCCESS(f'Connecté au broker ({reason_code})'))
+        client.subscribe(settings.MQTT_TOPIC_SUBSCRIBE)
+
+    def on_message(self, client: mqtt.Client, userdata: Any, message: mqtt.MQTTMessage) -> None:
+        try:
+            payload = json.loads(message.payload.decode('utf-8'))
+            save_ban_event(payload)
+        except Exception as e:
+            self.stderr.write(self.style.ERROR(f'Message MQTT invalide ({message.topic}) : {e}'))
+            return
+        self.stdout.write(f'Événement enregistré : {payload.get("name")} {payload.get("action")} {payload.get("ip")}')

emitter/banevents/management/commands/publish_command.py

@@ -0,0 +1,115 @@
+"""Déclenche à la main une commande sans émission automatique (Phase 4) :
+contrairement à SYNC_BAN/ESCALATE (toujours publiées par master_listen.py
+via self.correlation_rules), WHITELIST/BAN_ALLPORTS/NOTIFY_ONLY/RATE_LIMIT
+correspondent à une décision humaine ("je sais que cette IP est
+particulière"), pas à une règle automatique — pas de logique de
+corrélation à construire pour ça. Diffusées à tous les noeuds abonnés via
+fail2ban/broadcast/action, exécutées par master_client.py::execute_<nom>.
+
+REPORT_ABUSE fait exception : PAS de diffusion MQTT. Diffuser à chaque
+noeud produirait un rapport en double par noeud connecté pour le même
+incident (vu comme du bruit par AbuseIPDB, et demanderait une clé API par
+noeud). Ce process tourne déjà sur le master, avec accès à l'historique
+BanEvent complet (tous noeuds) pour construire un commentaire pertinent —
+il soumet donc le rapport lui-même (banevents/abuseipdb.py), sans passer
+par MQTT ni par master_client.py.
+
+Usage (à lancer sur le master, où vit le certificat master-internal, seule
+identité autorisée en écriture sur fail2ban/broadcast/action) :
+    manage.py publish_command WHITELIST 203.0.113.5
+    manage.py publish_command BAN_ALLPORTS 203.0.113.5
+    manage.py publish_command NOTIFY_ONLY 203.0.113.5
+    manage.py publish_command RATE_LIMIT 203.0.113.5
+    manage.py publish_command REPORT_ABUSE 203.0.113.5 --comment "texte"
+"""
+import json
+import ssl
+import time
+import uuid
+from typing import Any
+
+import paho.mqtt.client as mqtt
+from django.conf import settings
+from django.core.management.base import BaseCommand, CommandError
+
+from banevents.abuseipdb import submit_report
+from banevents.models import BanEvent
+
+BROADCAST_ACTION_TOPIC = 'fail2ban/broadcast/action'
+# Commandes diffusées en MQTT à tous les noeuds — cf. master_client.py
+# command_handlers pour la liste des commandes qu'un noeud sait exécuter.
+MANUAL_COMMANDS = ('WHITELIST', 'BAN_ALLPORTS', 'NOTIFY_ONLY', 'RATE_LIMIT')
+DEFAULT_ABUSEIPDB_CATEGORIES = '18'  # Brute-Force
+
+
+class Command(BaseCommand):
+    help = 'Publie une commande manuelle (WHITELIST/BAN_ALLPORTS/NOTIFY_ONLY/RATE_LIMIT/REPORT_ABUSE).'
+
+    def add_arguments(self, parser: Any) -> None:
+        parser.add_argument('cmd', choices=(*MANUAL_COMMANDS, 'REPORT_ABUSE'))
+        parser.add_argument('ip')
+        parser.add_argument('--categories', default=DEFAULT_ABUSEIPDB_CATEGORIES, help='REPORT_ABUSE uniquement')
+        parser.add_argument('--comment', default='', help='REPORT_ABUSE uniquement')
+
+    def handle(self, *args: Any, **options: Any) -> None:
+        cmd = options['cmd']
+        ip = options['ip']
+
+        if cmd == 'REPORT_ABUSE':
+            self.handle_report_abuse(ip, options['categories'], options['comment'])
+            return
+
+        client = mqtt.Client(
+            mqtt.CallbackAPIVersion.VERSION2, client_id=f'publish-command-{uuid.uuid4().hex[:8]}'
+        )
+        # Même identité que master_listen.py (master-internal) — seule
+        # scopée en écriture sur fail2ban/broadcast/action, cf.
+        # master-acl.conf.
+        client.tls_set(
+            certfile=settings.MQTT_MASTER_LISTENER_CERT,
+            keyfile=settings.MQTT_MASTER_LISTENER_KEY,
+            tls_version=ssl.PROTOCOL_TLSv1_2,
+        )
+        client.connect(settings.MQTT_MASTER_HOST, settings.MQTT_MASTER_PORT, keepalive=10)
+        client.loop_start()
+
+        payload = {'cmd': cmd, 'ip': ip, 'origin_node': 'manual', 'reason': 'manual'}
+        info = client.publish(BROADCAST_ACTION_TOPIC, json.dumps(payload), qos=1)
+        info.wait_for_publish(timeout=10)
+
+        client.loop_stop()
+        client.disconnect()
+
+        if not info.is_published():
+            raise CommandError(f'Échec de publication de {cmd} pour {ip} (PUBACK non reçu).')
+        self.stdout.write(self.style.SUCCESS(f'{cmd} publié pour {ip} sur {BROADCAST_ACTION_TOPIC}'))
+        # Laisse le temps au disconnect propre de partir avant que le
+        # process ne se termine (paho ferme le socket en tâche de fond).
+        time.sleep(0.2)
+
+    def handle_report_abuse(self, ip: str, categories: str, comment: str) -> None:
+        comment = comment or self._default_comment(ip)
+
+        if not settings.ABUSEIPDB_ENABLED or not settings.ABUSEIPDB_API_KEY:
+            self.stdout.write(self.style.WARNING(
+                f'[dry-run] AbuseIPDB désactivé (ABUSEIPDB_ENABLED/ABUSEIPDB_API_KEY, voir .env) — '
+                f'rien envoyé. Aurait signalé {ip} (catégories {categories}) : {comment}'
+            ))
+            return
+
+        ok, detail = submit_report(ip, categories, comment)
+        if ok:
+            self.stdout.write(self.style.SUCCESS(f'{ip} signalé à AbuseIPDB (catégories {categories}).'))
+        else:
+            raise CommandError(f'Échec du signalement AbuseIPDB pour {ip} : {detail}')
+
+    def _default_comment(self, ip: str) -> str:
+        """Commentaire par défaut construit depuis l'historique BanEvent
+        de cette IP — le master voit tous les noeuds, contrairement à un
+        rapport qui partirait d'un seul d'entre eux."""
+        events = BanEvent.objects.filter(ip_address=ip, action='BAN').order_by('-received_at')[:5]
+        jails = sorted({e.jail_name for e in events if e.jail_name})
+        nodes = sorted({e.node for e in events if e.node})
+        if not jails:
+            return f'Fail2Ban: banned IP {ip}.'
+        return f'Fail2Ban: banned for {", ".join(jails)} on {len(nodes)} node(s).'

emitter/banevents/management/commands/retry_geolocation.py

@@ -0,0 +1,39 @@
+"""Relance la géolocalisation pour des BanEvent (Phase 4 — repurposing de
+REQUEST_GEOLOCATE). La géoloc est déjà automatique et centralisée à
+l'ingestion (signal post_save sur BanEvent, cf. signals.py, déclenché pour
+tout événement y compris ceux ingérés par master_listen depuis n'importe
+quel noeud) : un aller-retour MQTT "demande au noeud sa géoloc" n'aurait
+rien à demander, le noeud fail2ban ne calcule rien lui-même. Cette
+commande se contente de rejouer les échecs (ip-api.com en rate limit, IP
+privée/invalide, ...) via le même Celery task, sans nouvelle logique.
+
+Usage :
+    manage.py retry_geolocation                  # tous les BanEvent sans coordonnées
+    manage.py retry_geolocation --ip 203.0.113.5  # une IP précise, même déjà tentée
+"""
+from typing import Any
+
+from django.core.management.base import BaseCommand
+
+from banevents.models import BanEvent
+from banevents.tasks import geolocate_ip
+
+
+class Command(BaseCommand):
+    help = 'Relance la géolocalisation (Celery) pour les BanEvent sans coordonnées, ou une IP précise.'
+
+    def add_arguments(self, parser: Any) -> None:
+        parser.add_argument('--ip', default='', help='Ne relancer que pour cette IP (même si déjà géolocalisée)')
+
+    def handle(self, *args: Any, **options: Any) -> None:
+        qs = BanEvent.objects.all()
+        if options['ip']:
+            qs = qs.filter(ip_address=options['ip'])
+        else:
+            qs = qs.filter(geolocated_at__isnull=True)
+
+        count = 0
+        for event_id in qs.values_list('id', flat=True):
+            geolocate_ip.delay(event_id)
+            count += 1
+        self.stdout.write(self.style.SUCCESS(f'{count} géolocalisation(s) relancée(s).'))