Le composant Rooter est un élément de sécurité fondamental au sein de l'écosystème Cuckoo Sandbox. Sa mission principale est d'exécuter en toute sécurité des opérations réseau nécessitant des privilèges root, et ce, dans un environnement non privilégié. En tant que module central de gestion réseau pour les systèmes automatisés d'analyse dynamique de logiciels malveillants, Rooter assure une communication sécurisée et une séparation des privilèges via les sockets UNIX, offrant un contrôle flexible du routage réseau pour chaque tâche d'analyse.
Architecture Fondamentale et Principes de Fonctionnement
Conception Architecturale
Rooter est conçu selon une architecture client-serveur, facilitant la communication inter-processus par le biais de sockets de domaine UNIX.
+---------------------+ +---------------------+
| Processus Cuckoo | | Composant Rooter |
| (Client non root) | | (Serveur root) |
+---------------------+ +---------------------+
| | ^ ^
| | | |
| +---- Commande JSON --->---------+
| |
+<---- Réponse JSON --------------+
| |
| |
(UNIX Domain Socket: /tmp/cuckoo-rooter)
Protocole de Communication
La communication avec Rooter s'effectue via un protocole simple basé sur JSON, encapsulant les commandes et leurs arguments.
{
"action": "activer_transfert",
"parametres": ["vboxnet0", "eth0", "192.168.56.101"],
"options_supp": {}
}
Fonctionnalités Clés
1. Gestion des Interfaces Réseau
Vérifier la disponibilité d'une interface réseau est crucial avant de lui assigner des règles. La fonction suivante permet de valider son existence et sa syntaxe.
import subprocess
import re
def valider_interface_reseau(nom_interface: str) -> bool:
"""
Vérifie si une interface réseau existe et si son nom est valide.
"""
if not re.match(r"^[a-zA-Z0-9-._]+$", nom_interface):
print(f"Nom d'interface invalide: {nom_interface}")
return False
try:
# Utilise 'ip link show' pour confirmer l'existence de l'interface
subprocess.run(
["ip", "link", "show", nom_interface],
check=True,
capture_output=True,
text=True
)
return True
except subprocess.CalledProcessError:
print(f"L'interface {nom_interface} n'existe pas ou n'est pas disponible.")
return False
2. Contrôle des Tables de Routage
Il est souvent nécessaire de vérifier l'existence d'une table de routage personnalisée pour les stratégies de routage avancées.
import subprocess
def verifier_table_routage_definie(nom_table_rt: str) -> bool:
"""
Détermine si une table de routage spécifique est configurée dans le système.
"""
try:
# Tente de lister les routes associées à la table
subprocess.run(
["ip", "route", "list", "table", nom_table_rt],
check=True,
capture_output=True,
text=True
)
return True
except subprocess.CalledProcessError:
print(f"La table de routage '{nom_table_rt}' n'est pas définie.")
return False
3. Statut des Services de Tunnel
Le Rooter peut gérer divers services de tunnel. Il est essentiel de pouvoir en interroger le statut.
import re
import subprocess
# Fonction utilitaire simplifiée pour l'exécution de commandes shell
# Dans un environnement Cuckoo réel, 's.service' serait un utilitaire système.
def executer_commande_systeme(*args):
try:
result = subprocess.run(
list(args),
check=True,
capture_output=True,
text=True,
encoding="utf-8"
)
return result.stdout.strip()
except subprocess.CalledProcessError:
return ""
def obtenir_etat_passerelles_tunnelees() -> dict:
"""
Récupère l'état (actif ou inactif) des passerelles de tunnel configurées.
"""
etats_tunnels = {}
# Exemple de sortie simulée pour un service de gestion de tunnels
# Une commande réelle pourrait être "systemctl status cuckoo-tunnels" ou similaire.
service_output = executer_commande_systeme("service", "network-tunnel", "status") # Hypothétique commande
for line_entry in service_output.split("\n"):
# Rechercher des motifs tels que "'mon_tunnel' est actif" ou "'autre_tunnel' est inactif"
match_obj = re.search(r"'(?P<nom_tunnel>\w+)' est (?P<inactif>non )?actif", line_entry)
if match_obj:
id_tunnel = match_obj.group("nom_tunnel")
est_actif = match_obj.group("inactif") is None # Si "non " n'est pas présent, le tunnel est actif
etats_tunnels[id_tunnel] = est_actif
return etats_tunnels
</inactif></nom_tunnel>
4. Translation d'Adresse Réseau (NAT)
L'activation du NAT est une opération courante pour permettre aux machines virtuelles d'accéder au réseau externe.
import subprocess
def configurer_nat_masquerade(interface_sortante: str):
"""
Ajoute une règle iptables pour activer le NAT (Masquerading)
sur une interface de sortie spécifiée.
"""
try:
subprocess.run(
["iptables", "-t", "nat", "-A", "POSTROUTING", "-o", interface_sortante, "-j", "MASQUERADE"],
check=True,
capture_output=True,
text=True
)
print(f"NAT Masquerading activé sur l'interface {interface_sortante}.")
except subprocess.CalledProcessError as e:
print(f"Échec de l'activation du NAT sur {interface_sortante}: {e.stderr.strip()}")
Modes de Routage Détaillés
Comparaison des Stratégies de Routage
| Mode de Routage | Description | Scénarios Applicables | Niveau de Sécurité |
|---|---|---|---|
| aucun | Pas de configuration de routage spécifique. | Environnements de test basiques. | Faible |
| bloquer | Interruption complète du trafic non initié par Cuckoo. | Analyse hautement isolée. | Élevé |
| internet | Accès complet au réseau internet. | Surveillance du comportement en ligne. | Moyen |
| inetsim | Redirection vers le simulateur de services réseau InetSim. | Simulation de services. | Élevé |
| tor | Routage du trafic via le réseau Tor pour l'anonymisation. | Analyse anonyme. | Moyen |
| tunnel | Routage via des tunnels VPN ou SSH configurés. | Tests de géolocalisation ou d'infrastructure spécifique. | Moyen |
Processus de Fonctionnement des Modes de Routage
Chaque mode de routage adapte dynamiquement les règles de pare-feu et les tables de routage du système hôte. Par exemple, le mode bloquer installe des règles iptables restrictives pour isoler la machine virtuelle, tandis que le mode internet configure des règles plus permissives et assure le NAT vers l'interface externe. Les modes inetsim et tor mettent en place des redirections de trafic (via DNAT/REDIRECT) vers leurs services respectifs, et le mode tunnel utilise des règles de politique de routage (ip rule) pour forcer le trafic d'une VM spécifique à travers une interface tunelée.
Installation et Configuration
Prérequis Système
- Système d'exploitation Linux (Ubuntu ou Debian recommandé).
- Accès root ou privilèges sudo.
- Suite d'outils
iptables. - Paquet
iproute2. - Services de tunnel réseau (si le mode
tunnelest utilisé).
Procédure d'Installation
-
Installation des dépendances: ```bash
sudo apt update sudo apt install iptables iproute2 network-manager # network-manager pour network-utils
-
Définition des tables de routage iproute2: Modifiez le fichier
/etc/iproute2/rt_tablespour y ajouter vos tables de routage personnalisées. Elles sont essentielles pour la politique de routage.# Valeurs réservées 255 local 254 main 253 default 0 unspec # Tables personnalisées pour Cuckoo 400 tun_us_east 401 tun_eu_west 402 tor_node_a -
Activation du transfert IP: Permettez au noyau Linux de transférer les paquets IP entre les interfaces.
echo 1 | sudo tee -a /proc/sys/net/ipv4/ip_forward > /dev/null sudo sysctl -w net.ipv4.ip_forward=1
Démarrage du Composant Rooter
Le Rooter doit être exécuté avec des privilèges root, mais il peut être démarré de différentes manières, en s'assurant que le processus Cuckoo non privilégié peut communiquer avec lui.
Méthode 1: Exécution directe en root
sudo cuckoo rooter -g groupe_cuckoo
Méthode 2: Utilisation d'un environnement virtuel (virtualenv)
sudo /chemin/vers/venv/bin/cuckoo rooter -g votre_utilisateur
Méthode 3: Avec le drapeau sudo (Cuckoo gère les privilèges)
cuckoo rooter --sudo -g votre_utilisateur
Exemples de Configuration Avancée
Configuration du Routage par Tunnel
Définissez plusieurs tunnels avec leurs interfaces et tables de routage associées dans le fichier de configuration routing.conf.
; $CWD/conf/routing.conf
[tunnel]
enabled = yes
tunnels = tun_us_east, tun_eu_west
[tun_us_east]
name = tunnel_usa_est
description = Côte Est des États-Unis
interface = tun0
rt_table = tun_us_east
[tun_eu_west]
name = tunnel_europe_ouest
description = Allemagne, Europe
interface = tun1
rt_table = tun_eu_west
Configuration du Routage Tor
Pour acheminer le trafic via Tor, configurez les ports DNS et proxy dans routing.conf.
; $CWD/conf/routing.conf
[tor]
enabled = yes
dns_port = 5353
proxy_port = 9040
Diagnostic et Dépannage
Résolution des Problèmes Courants
-
Questions de permissions: ```bash
Vérifier les permissions du fichier socket UNIX
ls -la /tmp/cuckoo-rooter
Permissions attendues: srw-rw---- 1 root cuckoo
-
Problèmes de chemin des exécutables: Assurez-vous que Cuckoo peut trouver les outils système (
service,iptables,ip).cuckoo rooter --service /usr/sbin/service --iptables /sbin/iptables --ip /sbin/ip -
Anomalies d'interface réseau: ```bash
Lister l'état des interfaces
ip link show
Examiner les tables de routage
ip route list table main
Mode Débogage
Activez la journalisation détaillée pour obtenir des informations de diagnostic supplémentaires.
cuckoo rooter --debug 2>&1 | tee rooter.log
Bonnes Pratiques de Sécurité
1. Principe du Moindre Privilège
- Exécutez Rooter avec un utilisateur dédié et aux privilèges minimaux.
- Restreignez l'accès au fichier socket du Rooter.
- Examinez et mettez à jour régulièrement les règles iptables.
2. Isolation Réseau
Appliquez des règles iptables par défaut pour restreindre le trafic non autorisé.
# Bloquer tout transfert par défaut
sudo iptables -P FORWARD DROP
# Autoriser uniquement les connexions établies et liées
sudo iptables -A FORWARD -m state --state RELATED,ESTABLISHED -j ACCEPT
3. Surveillance et Audit
Surveillez l'activité du Rooter et auditez les modifications de pare-feu.
# Vérifier les connexions actives au socket du Rooter
sudo netstat -an | grep cuckoo-rooter
# Sauvegarder les règles iptables pour l'audit
sudo iptables-save > /var/log/iptables_audit_$(date +%Y%m%d).txt
Optimisation des Performances
1. Optimisation du Pool de Connexions
Réutilisez les connexions aux sockets UNIX pour minimiser la surcharge de création.
import threading
import queue
from typing import Optional
# Une classe factice pour représenter une connexion Rooter
class ConnexionRooter:
def __init__(self, id_conn: int):
self.id = id_conn
# Ici, la vraie logique d'établissement de socket UNIX
# print(f"Connexion Rooter #{id_conn} établie.")
def envoyer(self, data: dict) -> dict:
# Simule l'envoi de données et la réception d'une réponse
# print(f"Connexion #{self.id}: Envoi de {data['commande']}")
return {"statut": "succès", "message": f"Commande {data['commande']} exécutée"}
def fermer(self):
# print(f"Connexion Rooter #{self.id} fermée.")
pass
class GestionnairePoolConnexionsRooter:
"""
Gère un pool de connexions Rooter pour une meilleure efficacité.
"""
def __init__(self, taille_max_pool: int = 5):
self._pool_libre: queue.Queue[ConnexionRooter] = queue.Queue(maxsize=taille_max_pool)
self._taille_max = taille_max_pool
self._compteur_connexions = 0
self._lock_creation = threading.Lock()
def _creer_nouvelle_connexion(self) -> ConnexionRooter:
with self._lock_creation:
self._compteur_connexions += 1
return ConnexionRooter(self._compteur_connexions)
def obtenir_connexion(self) -> ConnexionRooter:
"""
Récupère une connexion existante du pool ou en crée une nouvelle.
"""
if not self._pool_libre.empty():
return self._pool_libre.get_nowait()
else:
return self._creer_nouvelle_connexion()
def liberer_connexion(self, connexion: ConnexionRooter):
"""
Remet une connexion utilisée dans le pool. Si le pool est plein, la ferme.
"""
try:
self._pool_libre.put_nowait(connexion)
except queue.Full:
connexion.fermer()
with self._lock_creation:
self._compteur_connexions -= 1
def fermer_tout(self):
""" Ferme toutes les connexions restantes dans le pool. """
while not self._pool_libre.empty():
conn = self._pool_libre.get_nowait()
conn.fermer()
with self._lock_creation:
self._compteur_connexions -= 1
2. Prise en Charge des Opérations par Lot
Regroupez plusieurs commandes réseau pour les envoyer en une seule requête, réduisant ainsi la latence.
from typing import List, Dict, Any, Callable
# Table de dispatching des gestionnaires de commandes (exemple)
_gestionnaires_commandes: Dict[str, Callable[..., Any]] = {
"add_rule": lambda chain, rule: print(f"Ajout de la règle {rule} à {chain}"),
"remove_rule": lambda chain, rule: print(f"Suppression de la règle {rule} de {chain}"),
"set_nat": lambda interface: print(f"Configuration NAT pour {interface}")
}
def executer_operations_en_lot(liste_commandes: List[Dict[str, Any]]):
"""
Exécute une liste d'opérations réseau fournies sous forme de lot.
Chaque élément du lot est un dictionnaire décrivant une commande.
"""
for commande_spec in liste_commandes:
nom_cmd = commande_spec.get("nom_commande")
args_cmd = commande_spec.get("arguments", [])
kwargs_cmd = commande_spec.get("options", {})
if nom_cmd and nom_cmd in _gestionnaires_commandes:
try:
_gestionnaires_commandes[nom_cmd](*args_cmd, **kwargs_cmd)
# print(f"Commande '{nom_cmd}' exécutée avec succès.")
except Exception as e:
print(f"Erreur lors de l'exécution de '{nom_cmd}': {e}")
else:
print(f"Commande inconnue ou mal formée: '{nom_cmd}'")
3. Traitement Asynchrone
Utilisez un thread séparé pour exécuter les opérations Rooter, évitant de bloquer le thread principal de Cuckoo.
import threading
import queue
import time
from typing import Dict, Any
# Réutilisons les gestionnaires de commandes pour l'exemple asynchrone
_gestionnaires_commandes_async: Dict[str, Callable[..., Any]] = {
"appliquer_routage": lambda vm_ip, gw_ip: print(f"[ASYNC] Routage VM {vm_ip} via {gw_ip} configuré."),
"mettre_a_jour_parefeu": lambda action, param: print(f"[ASYNC] Pare-feu: {action} {param}."),
"delai_simule": lambda sec: time.sleep(sec) or print(f"[ASYNC] Délai de {sec}s terminé.")
}
class OrchestrateurRooterAsynchrone:
"""
Permet l'exécution non bloquante des commandes Rooter via une file d'attente
et un thread de travail dédié.
"""
def __init__(self):
self._file_taches: queue.Queue[Dict[str, Any]] = queue.Queue()
self._thread_executeur = threading.Thread(target=self._boucle_execution_taches)
self._thread_executeur.daemon = True # Permet au programme principal de se terminer
self._actif = False
def demarrer(self):
""" Démarre le thread de travail asynchrone. """
if not self._actif:
self._actif = True
self._thread_executeur.start()
print("Orchestrateur Rooter asynchrone démarré.")
def arreter(self):
""" Arrête le thread de travail et attend sa terminaison. """
if self._actif:
self._actif = False
self._thread_executeur.join()
print("Orchestrateur Rooter asynchrone arrêté.")
def _boucle_execution_taches(self):
""" Boucle principale du thread de travail pour traiter les tâches. """
while self._actif:
try:
tache = self._file_taches.get(timeout=0.5) # Attente avec timeout
self._traiter_tache(tache)
self._file_taches.task_done()
except queue.Empty:
continue # Continuer la boucle si la file est vide
def _traiter_tache(self, tache_courante: Dict[str, Any]):
""" Exécute une tâche Rooter individuelle. """
cmd_name = tache_courante.get("commande")
cmd_args = tache_courante.get("args", [])
cmd_kwargs = tache_courante.get("kwargs", {})
if cmd_name and cmd_name in _gestionnaires_commandes_async:
try:
_gestionnaires_commandes_async[cmd_name](*cmd_args, **cmd_kwargs)
except Exception as e:
print(f"Erreur d'exécution asynchrone pour '{cmd_name}': {e}")
else:
print(f"Commande asynchrone inconnue: '{cmd_name}'")
def ajouter_tache(self, commande: str, args: list = None, kwargs: dict = None):
""" Ajoute une commande à la file d'attente pour une exécution asynchrone. """
self._file_taches.put({"commande": commande, "args": args or [], "kwargs": kwargs or {}})
print(f"Tâche '{commande}' ajoutée à la file d'attente.")
Extension et Personnalisation
Implémentation de Gestionnaires de Routage Personnalisés
Vous pouvez étendre les capacités du Rooter en définissant vos propres fonctions de routage.
import subprocess
from typing import List
# Fonctions utilitaires pour exécuter des commandes système
def _appliquer_iptables(*args: str):
subprocess.run(["iptables"] + list(args), check=True, capture_output=True, text=True)
def _executer_ip_cmd(*args: str):
subprocess.run(["ip"] + list(args), check=True, capture_output=True, text=True)
def configurer_itineraire_specifique_vm(ip_vm: str, passerelle: str, interface_locale: str):
"""
Met en place une configuration de routage et de pare-feu spécifique
pour une machine virtuelle donnée.
"""
print(f"Configuration d'itinéraire spécifique pour {ip_vm} via {passerelle} sur {interface_locale}.")
# Autoriser le trafic de la VM vers l'interface locale et vice-versa
_appliquer_iptables("-A", "FORWARD", "-s", ip_vm, "-o", interface_locale, "-j", "ACCEPT")
_appliquer_iptables("-A", "FORWARD", "-d", ip_vm, "-i", interface_locale, "-j", "ACCEPT")
# Définir une règle de politique de routage: le trafic provenant de l'IP de la VM
# utilisera une table de routage personnalisée.
nom_table_perso = f"vm_route_{ip_vm.replace('.', '_')}"
_executer_ip_cmd("rule", "add", "from", ip_vm, "table", nom_table_perso)
# Ajouter la route par défaut dans cette table personnalisée
_executer_ip_cmd("route", "add", "default", "via", passerelle, "table", nom_table_perso)
# Note: La table de routage 'nom_table_perso' doit être déclarée dans /etc/iproute2/rt_tables
# (par ex. "500 vm_route_xxx") pour être reconnue par le système.
Intégration de l'Architecture de Plugins
Le Rooter peut être étendu en enregistrant de nouveaux gestionnaires de commandes dans sa table de dispatching.
# Table de dispatching des commandes principales du Rooter
table_dispatch_rooter_principale = {
"configurer_base_nat": lambda iface: print(f"NAT de base configuré sur {iface}"),
"gerer_dns": lambda vm_ip, dns_ip: print(f"DNS pour {vm_ip} défini sur {dns_ip}"),
# ... autres gestionnaires de commandes internes ...
}
# Fonctionnalités personnalisées à intégrer
def gerer_nat_complex(interface_in: str, interface_out: str, port_map: dict):
print(f"Gestion NAT complexe: {interface_in} -> {interface_out} avec {port_map}")
def enregistrer_extensions_reseau():
"""
Ajoute des fonctions de gestion réseau personnalisées à la table
de dispatching principale du Rooter.
"""
print("Enregistrement de nouvelles commandes réseau dans le Rooter...")
table_dispatch_rooter_principale.update({
"configurer_itineraire_vm": configurer_itineraire_specifique_vm,
"gerer_nat_avance": gerer_nat_complex,
})
print(f"Nouvelles commandes disponibles: {list(table_dispatch_rooter_principale.keys())}")
# Exemple d'appel après enregistrement
# enregistrer_extensions_reseau()
# table_dispatch_rooter_principale["configurer_itineraire_vm"]("192.168.1.100", "192.168.1.1", "eth0")