from typing import TYPE_CHECKING from django.db import models from django.utils import timezone class BanEvent(models.Model): """Un événement de ban/notice reçu via MQTT depuis un noeud fail2ban.""" node = models.CharField(max_length=64) action = models.CharField(max_length=32) jail_name = models.CharField(max_length=64, db_index=True) ip_address = models.GenericIPAddressField() # Chaîne, pas un entier : fail2ban accepte des noms de service # ("http,https") ou des listes de ports ("8883,8884") pour ce champ. port = models.CharField(max_length=64, blank=True, default='') protocol = models.CharField(max_length=16, blank=True) bantime = models.IntegerField(null=True, blank=True) reason = models.CharField(max_length=255, blank=True) event_time = models.DateTimeField(null=True, blank=True) received_at = models.DateTimeField(auto_now_add=True) raw_payload = models.JSONField(default=dict, blank=True) latitude = models.FloatField(null=True, blank=True) longitude = models.FloatField(null=True, blank=True) country = models.CharField(max_length=64, blank=True, db_index=True) country_code = models.CharField(max_length=2, blank=True) city = models.CharField(max_length=128, blank=True) geolocated_at = models.DateTimeField(null=True, blank=True) class Meta: ordering = ['-received_at'] if TYPE_CHECKING: # Attaché dynamiquement par views.py::_attach_display_node (alias # résolu depuis NodeRegistry), pas un champ persisté — annotation # purement pour le typage, aucun effet à l'exécution. display_node: str def __str__(self) -> str: return f'{self.jail_name} {self.action} {self.ip_address}' class NodeRegistry(models.Model): """Métadonnées par noeud (alias, pays du VPS), en plus du `node` brut (uuid.getnode() stringifié) porté par chaque BanEvent. Alimenté automatiquement par banevents.ingestion.save_ban_event dès qu'un noeud inédit publie un événement ; alias/country restent vides jusqu'à édition manuelle via /admin/ (seule UI d'édition, pas de page dédiée).""" node_id = models.CharField(max_length=64, unique=True) alias = models.CharField(max_length=100, blank=True) # Pays du VPS/noeud lui-même (saisi à la main) — à ne pas confondre avec # BanEvent.country, le pays géolocalisé de l'IP *bannie*, un concept # différent qui existe déjà indépendamment de ce modèle. country = models.CharField(max_length=100, blank=True) # URL de base du dashboard de CE noeud (ex. https://banishing.example.org, # sans slash final), saisie à la main via /admin/ — republiée à tous les # noeuds par le roster (master_listen.py) pour que le bloc "Noeuds" de # la sidebar (n'importe où il y a plus d'un noeud connu) puisse pointer # vers chacun sans maillage N×N codé en dur (cf. MASTER_DASHBOARD_URL, # qui reste le seul lien client -> master). dashboard_url = models.URLField(max_length=255, blank=True) first_seen = models.DateTimeField(auto_now_add=True) last_seen = models.DateTimeField(auto_now=True) class Meta: ordering = ['node_id'] if TYPE_CHECKING: # Attaché dynamiquement par views.py::_node_choices (calculé depuis # last_seen, pas persisté) — annotation purement pour le typage. is_online: bool def __str__(self) -> str: return self.alias or self.node_id @property def display_name(self) -> str: return self.alias or self.node_id class JoinToken(models.Model): """Jeton à usage unique pour l'auto-inscription d'un noeud (flux "kubeadm join", voir manage.py create_join_token et banevents/views.py::join_node). Seul le hash est stocké — jamais le jeton en clair, même logique qu'un hachage de mot de passe : une fuite de la base ne rend pas les jetons utilisables.""" node_name = models.CharField(max_length=64) token_hash = models.CharField(max_length=64, unique=True) created_at = models.DateTimeField(auto_now_add=True) expires_at = models.DateTimeField() used_at = models.DateTimeField(null=True, blank=True) class Meta: ordering = ['-created_at'] def __str__(self) -> str: if self.used_at: status = 'utilisé' elif timezone.now() > self.expires_at: status = 'expiré' else: status = 'actif' return f'{self.node_name} ({status})' class EnrollmentRequest(models.Model): """Demande d'inscription à la communauté, soumise via la page publique /communaute/ (désactivée par défaut, cf. COMMUNITY_ENROLLMENT_ENABLED). Ne crée jamais directement d'entrée NodeRegistry : le node_id réel (uuid.getnode() du futur noeud) n'existe pas encore à ce stade, il n'apparaît qu'au premier événement/heartbeat une fois le noeud effectivement rattaché — NodeRegistry reste le résultat final du parcours, pas une action immédiate de cette demande. Validée à la main dans /admin/ (voir EnrollmentRequestAdmin.approve_and_send_token) : jamais de jeton émis/envoyé sans décision humaine explicite.""" STATUS_PENDING = 'pending' STATUS_APPROVED = 'approved' STATUS_REJECTED = 'rejected' STATUS_CHOICES = [ (STATUS_PENDING, 'En attente'), (STATUS_APPROVED, 'Approuvée'), (STATUS_REJECTED, 'Rejetée'), ] email = models.EmailField() node_name = models.CharField(max_length=64) # Pays du VPS/serveur candidat (pas de l'IP d'un attaquant) — signalé # comme utile en pratique : les data-centers de pays différents ne # font pas face au même trafic malveillant, une information qu'on ne # peut pas déduire automatiquement à l'inscription. Propagé vers # NodeRegistry.country à la création (ingestion.save_ban_event), sur # le même principe que node_name -> NodeRegistry.alias. country = models.CharField(max_length=100, blank=True) message = models.TextField(blank=True) status = models.CharField(max_length=16, choices=STATUS_CHOICES, default=STATUS_PENDING) created_at = models.DateTimeField(auto_now_add=True) processed_at = models.DateTimeField(null=True, blank=True) class Meta: ordering = ['-created_at'] def __str__(self) -> str: # dict(...)[...] plutôt que get_status_display() : cette méthode # auto-générée par Django (choices=) n'est visible de pyright # qu'avec le plugin mypy de django-stubs, non supporté ici (même # limitation que Model.pk vs .id, voir CONTEXT.md/ROADMAP.md). return f'{self.email} ({self.node_name}) — {dict(self.STATUS_CHOICES)[self.status]}'