Fail2banMqttActionBanishment/emitter/banevents/views.py

import datetime
import hashlib
import json
import logging
import subprocess
import tempfile
from pathlib import Path

import redis
from django.conf import settings
from django.contrib import messages
from django.core.mail import mail_admins
from django.core.serializers.json import DjangoJSONEncoder
from django.http import Http404, HttpRequest, HttpResponse, JsonResponse
from django.shortcuts import redirect, render
from django.utils import timezone
from django.utils.translation import gettext as _
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from ipware import get_client_ip

from .forms import EnrollmentRequestForm
from .models import BanEvent, JoinToken, NodeRegistry
from .serializers import event_payload
from .stats import ROSTER_REDIS_DB, ROSTER_REDIS_KEY, top_jails_and_countries

join_logger = logging.getLogger('banevents.join')
SCRIPTS_DIR = Path(settings.BASE_DIR).parent / 'scripts'

HISTORY_MAX_EVENTS = 500
# 3x l'intervalle de heartbeat (HEARTBEAT_INTERVAL_SECONDS dans
# master_client.py) : tolère un accroc réseau/reconnexion sans faux
# positif "hors ligne".
NODE_OFFLINE_THRESHOLD_SECONDS = 180


def _node_choices() -> list[NodeRegistry]:
    """Attache `is_online` (calculé, pas un champ modèle) à chaque noeud —
    un noeud pas encore redéployé avec le heartbeat (Phase 3) apparaîtra
    "hors ligne" même s'il fonctionne, jusqu'à son prochain déploiement."""
    threshold = timezone.now() - datetime.timedelta(seconds=NODE_OFFLINE_THRESHOLD_SECONDS)
    nodes = list(NodeRegistry.objects.all())
    for node in nodes:
        node.is_online = node.last_seen >= threshold
    return nodes


def _filterable_nodes(nodes: list[NodeRegistry]) -> list[NodeRegistry]:
    """Sous-ensemble de `nodes` ayant au moins un BanEvent en local —
    depuis le roster (Phase 3), NodeRegistry liste TOUS les noeuds connus
    du master, y compris ceux dont ce process-ci n'a ingéré aucun
    événement (un noeud client ne voit localement que ses propres bans).
    Le menu de filtre ne doit proposer que des choix qui renvoient
    vraiment quelque chose — la liste "Noeuds" de la sidebar, elle,
    continue d'afficher tout le monde (c'est son rôle)."""
    local_node_ids = set(BanEvent.objects.order_by().values_list('node', flat=True).distinct())
    return [node for node in nodes if node.node_id in local_node_ids]


def _attach_display_node(events: list[BanEvent], nodes: list[NodeRegistry]) -> None:
    """Résout l'alias de chaque event en une seule requête (déjà chargée
    dans `nodes`) plutôt qu'un lookup par event — attaché comme attribut
    Python, lu à la fois par le template (event.display_node) et par
    event_payload() pour le JSON de bootstrap."""
    alias_map = {node.node_id: node.display_name for node in nodes}
    for event in events:
        event.display_node = str(alias_map.get(event.node, event.node))


def _sidebar_stats() -> dict[str, list[dict]]:
    """Compteurs jail/pays pour le widget de la sidebar — volontairement
    pas filtrés par le ?node=/plage de dates de la page courante, cf.
    plan : c'est un résumé permanent, pas du contenu de page.

    Essaie d'abord le roster reçu du master (Phase 3, cache Redis alimenté
    par master_client.py) : sur un noeud client, les BanEvent locaux ne
    couvrent que ce noeud, la vue globale vient forcément d'ailleurs. Repli
    sur le calcul local si absent/périmé — vrai sur le master lui-même
    (qui republie ce qu'il vient de calculer localement, donc aucune
    différence pratique) ou sur un noeud pas encore redéployé avec le
    roster."""
    try:
        cached = redis.Redis(host=settings.REDIS_HOST, port=settings.REDIS_PORT, db=ROSTER_REDIS_DB).get(
            ROSTER_REDIS_KEY
        )
    except redis.RedisError:
        cached = None
    # isinstance, pas juste `if cached:` : redis-py type .get() de façon
    # générique (client sync ET async partagent la signature), pyright ne
    # peut pas savoir statiquement que CE client est synchrone — ce client
    # ne renvoie jamais un awaitable en pratique.
    if isinstance(cached, (str, bytes)):
        return json.loads(cached)
    return top_jails_and_countries()


def dashboard(request: HttpRequest) -> HttpResponse:
    selected_node = request.GET.get('node', '')

    events_qs = BanEvent.objects.all()
    if selected_node:
        events_qs = events_qs.filter(node=selected_node)
    events = list(events_qs[:100])

    nodes = _node_choices()
    _attach_display_node(events, nodes)

    events_json = json.dumps(
        [event_payload(event, event.display_node) for event in events], cls=DjangoJSONEncoder
    )
    return render(request, 'banevents/dashboard.html', {
        'events': events,
        'events_json': events_json,
        'nodes': nodes,
        'filterable_nodes': _filterable_nodes(nodes),
        'selected_node': selected_node,
        'master_dashboard_url': settings.MASTER_DASHBOARD_URL,
        'sidebar_width': settings.DASHBOARD_SIDEBAR_WIDTH,
        'community_enrollment_enabled': settings.COMMUNITY_ENROLLMENT_ENABLED,
        **_sidebar_stats(),
    })


def _join_client_ip(request: HttpRequest) -> str:
    """Même résolution d'IP réelle qu'axes (nginx en unique reverse-proxy,
    cf. AXES_IPWARE_PROXY_COUNT/AXES_IPWARE_META_PRECEDENCE_ORDER dans
    config/settings/base.py et son historique de bugs) — réutilisée ici
    uniquement pour l'audit (logs), pas pour une décision de sécurité."""
    ip, _ = get_client_ip(
        request,
        proxy_count=settings.AXES_IPWARE_PROXY_COUNT,
        request_header_order=settings.AXES_IPWARE_META_PRECEDENCE_ORDER,
    )
    return ip or '?'


def _join_error(request: HttpRequest, node_name: str, reason: str) -> JsonResponse:
    """Message volontairement générique côté client quelle que soit la
    cause réelle (jeton inconnu, expiré, déjà utilisé, CN invalide, échec
    de signature) — ne jamais donner d'indice à un tiers qui sonderait cet
    endpoint (énumération de noms de noeuds, etc.). Le détail part
    uniquement dans les logs serveur."""
    join_logger.warning('Échec join pour %s depuis %s : %s', node_name, _join_client_ip(request), reason)
    return JsonResponse({'error': 'Jeton invalide ou expiré.'}, status=403)


@csrf_exempt
@require_POST
def join_node(request: HttpRequest) -> JsonResponse:
    """Auto-inscription d'un noeud par jeton à usage unique (flux
    "kubeadm join", voir manage.py create_join_token et ROADMAP.md).
    Appel machine-à-machine (sudo make join côté noeud), pas de
    session navigateur — csrf_exempt. Aucune jail fail2ban dédiée : les
    jetons ont 256 bits d'entropie (secrets.token_urlsafe(32)), un
    brute-force est déjà impraticable sans compteur d'échecs, contrairement
    à /admin/ (mots de passe, entropie humaine faible, d'où django-axes)."""
    try:
        payload = json.loads(request.body)
        node_name = str(payload['node_name'])
        token = str(payload['token'])
        csr_pem = str(payload['csr'])
    except (json.JSONDecodeError, KeyError, TypeError, UnicodeDecodeError):
        return _join_error(request, '?', 'requête JSON invalide')

    token_hash = hashlib.sha256(token.encode('utf-8')).hexdigest()
    now = timezone.now()
    # update() atomique : une seule ligne affectée garantit qu'aucune
    # requête concurrente n'a déjà consommé ce jeton (protection contre
    # une double utilisation en cas de rejeu/course). Si tout ce qui suit
    # échoue (CN incohérent, signature en erreur), le jeton est "rendu"
    # (used_at remis à NULL, cf. plus bas) plutôt que brûlé pour de bon —
    # un échec transitoire côté serveur ne doit pas forcer à réémettre un
    # jeton entièrement nouveau alors que le client n'y est pour rien.
    updated = JoinToken.objects.filter(
        token_hash=token_hash, node_name=node_name, used_at__isnull=True, expires_at__gt=now,
    ).update(used_at=now)
    if updated != 1:
        return _join_error(request, node_name, 'jeton inconnu, expiré ou déjà utilisé')

    with tempfile.TemporaryDirectory() as tmp_dir:
        csr_path = Path(tmp_dir) / 'node.csr'
        csr_path.write_text(csr_pem)

        # CN de la CSR doit correspondre au node_name déclaré : défense en
        # profondeur, un jeton (donc un node_name) ne permet pas de faire
        # signer un certificat pour un AUTRE nom. -nameopt oneline,-space_eq
        # force un format de sortie stable ("subject=CN=xxx", sans espace
        # autour du "="), indépendant de la version d'openssl installée.
        subject = subprocess.run(
            ['openssl', 'req', '-in', str(csr_path), '-noout', '-subject', '-nameopt', 'oneline,-space_eq'],
            capture_output=True, text=True, timeout=10,
        )
        if subject.returncode != 0 or subject.stdout.strip() != f'subject=CN={node_name}':
            JoinToken.objects.filter(token_hash=token_hash).update(used_at=None)
            return _join_error(request, node_name, f'CSR illisible ou CN incohérent ({subject.stdout.strip()!r})')

        result = subprocess.run(
            ['sudo', str(SCRIPTS_DIR / 'sign-node-csr.sh'), node_name, str(csr_path)],
            capture_output=True, text=True, timeout=15,
        )
        if result.returncode != 0:
            JoinToken.objects.filter(token_hash=token_hash).update(used_at=None)
            return _join_error(request, node_name, f'échec de signature : {result.stderr.strip()}')

    # ca.crt appartient à root:mosquitto, mode 640 (cf. master-ca-init.sh) —
    # le user de déploiement (celui qui fait tourner Daphne) n'est pas dans
    # ce groupe, une lecture directe échoue (PermissionError, repéré en
    # conditions réelles). sudo cat, comme pour la signature elle-même :
    # même accès déjà en place (NOPASSWD:ALL), pas de nouveau droit à ouvrir.
    ca_cert_result = subprocess.run(
        ['sudo', 'cat', str(Path(settings.MQTT_MASTER_CA_DIR) / 'ca.crt')],
        capture_output=True, text=True, timeout=10,
    )
    ca_cert_pem = ca_cert_result.stdout if ca_cert_result.returncode == 0 else ''

    join_logger.info('Noeud %s inscrit depuis %s', node_name, _join_client_ip(request))
    return JsonResponse({'cert': result.stdout, 'ca_cert': ca_cert_pem}, status=201)


def history(request: HttpRequest) -> HttpResponse:
    selected_node = request.GET.get('node', '')
    from_date = request.GET.get('from', '')
    to_date = request.GET.get('to', '')

    events_qs = BanEvent.objects.all()
    if selected_node:
        events_qs = events_qs.filter(node=selected_node)
    if from_date:
        events_qs = events_qs.filter(received_at__date__gte=from_date)
    if to_date:
        events_qs = events_qs.filter(received_at__date__lte=to_date)
    events = list(events_qs[:HISTORY_MAX_EVENTS])

    nodes = _node_choices()
    _attach_display_node(events, nodes)

    events_json = json.dumps(
        [event_payload(event, event.display_node) for event in events], cls=DjangoJSONEncoder
    )
    return render(request, 'banevents/history.html', {
        'events': events,
        'events_json': events_json,
        'nodes': nodes,
        'filterable_nodes': _filterable_nodes(nodes),
        'selected_node': selected_node,
        'from_date': from_date,
        'to_date': to_date,
        'master_dashboard_url': settings.MASTER_DASHBOARD_URL,
        'sidebar_width': settings.DASHBOARD_SIDEBAR_WIDTH,
        'community_enrollment_enabled': settings.COMMUNITY_ENROLLMENT_ENABLED,
        **_sidebar_stats(),
    })


def community_landing(request: HttpRequest) -> HttpResponse:
    """Page publique expliquant le projet et proposant de rejoindre la
    communauté (master uniquement, désactivée par défaut — voir
    COMMUNITY_ENROLLMENT_ENABLED). Ne crée jamais de jeton ni d'accès
    direct : la soumission place juste une EnrollmentRequest en attente,
    validée à la main dans /admin/ (voir
    EnrollmentRequestAdmin.approve_and_send_token)."""
    if not settings.COMMUNITY_ENROLLMENT_ENABLED:
        raise Http404

    if request.method == 'POST':
        form = EnrollmentRequestForm(request.POST)
        if form.is_valid():
            if form.is_spam():
                # Ignoré silencieusement : ne pas révéler à un bot qu'il a
                # été détecté (pas d'erreur, pas d'entrée créée).
                return redirect('banevents:community')
            enrollment = form.save()
            try:
                mail_admins(
                    'Nouvelle demande d\'inscription à la communauté',
                    f"Email : {enrollment.email}\n"
                    f"Noeud souhaité : {enrollment.node_name}\n"
                    f"Message : {enrollment.message or '(vide)'}\n\n"
                    f"À traiter dans /admin/banevents/enrollmentrequest/",
                )
            except Exception:
                # Best-effort : la demande est déjà enregistrée (visible dans
                # /admin/ de toute façon) — un échec d'envoi ne doit jamais
                # faire perdre la soumission de l'utilisateur.
                join_logger.exception('Échec de notification admin pour une nouvelle EnrollmentRequest')
            messages.success(request, _(
                'Votre demande est enregistrée. Vous recevrez un email si elle est acceptée.'
            ))
            return redirect('banevents:community')
    else:
        form = EnrollmentRequestForm()

    return render(request, 'banevents/community.html', {'form': form})