mirror of
https://github.com/deunix-educ/Fail2banMqttActionBanishment.git
synced 2026-08-24 03:11:58 +02:00
40 lines
1.7 KiB
Python
40 lines
1.7 KiB
Python
"""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).'))
|