Analyse des Canaux Cachés Réseau dans les Malwares
Aperçu
Les malwares utilisent des canaux cachés pour déguiser les communications C2 et l'exfiltration de données dans un trafic réseau légitime. Le tunneling DNS encode les données dans les requêtes et réponses DNS (utilisé par des outils comme iodine, dnscat2 et des familles de malwares comme FrameworkPOS). Le tunneling ICMP cache des données dans les payloads echo request/reply (icmpsh, ptunnel). Les canaux HTTP cachés intègrent les données C2 dans les en-têtes, les cookies ou les images stéganographiques. L'abus de protocole exploite les protocoles autorisés pour contourner les pare-feux. La détection du tunneling DNS atteint un rappel de 99 %+ avec les approches modernes basées sur l'apprentissage automatique, bien que l'exfiltration à faible débit reste difficile. Palo Alto Unit42 a suivi trois grandes campagnes de tunneling DNS (TrkCdn, SecShow, Savvy Seahorse) jusqu'en 2024, montrant la persistance de cette technique.
Quand l'utiliser
- Lors d'investigations d'incidents de sécurité nécessitant l'analyse des canaux cachés réseau dans les malwares
- Lors de la création de règles de détection ou de requêtes de threat hunting pour ce domaine
- Lorsque des analystes SOC ont besoin de procédures structurées pour ce type d'analyse
- Lors de la validation de la couverture de surveillance de sécurité pour les techniques d'attaque associées
Prérequis
- Python 3.9+ avec
scapy,dpkt,dnslib - Wireshark/tshark pour l'analyse PCAP
- Zeek (anciennement Bro) pour la surveillance réseau
- Infrastructure de journalisation des requêtes DNS
- Compréhension des protocoles DNS, ICMP, HTTP au niveau des paquets
Workflow
Étape 1 : Détection du Tunneling DNS
#!/usr/bin/env python3
"""Détecter le tunneling DNS et les canaux cachés dans le trafic réseau."""
import sys
import json
import math
from collections import Counter, defaultdict
try:
from scapy.all import rdpcap, DNS, DNSQR, DNSRR, IP, ICMP
except ImportError:
print("pip install scapy")
sys.exit(1)
def entropy(data):
if not data:
return 0
freq = Counter(data)
length = len(data)
return -sum((c/length) * math.log2(c/length) for c in freq.values())
def analyze_dns_tunneling(pcap_path):
"""Détecter les indicateurs de tunneling DNS dans un PCAP."""
packets = rdpcap(pcap_path)
domain_stats = defaultdict(lambda: {
"queries": 0, "total_qname_len": 0, "subdomain_lengths": [],
"query_types": Counter(), "unique_subdomains": set(),
})
for pkt in packets:
if pkt.haslayer(DNS) and pkt.haslayer(DNSQR):
qname = pkt[DNSQR].qname.decode('utf-8', errors='replace').rstrip('.')
qtype = pkt[DNSQR].qtype
parts = qname.split('.')
if len(parts) >= 3:
base_domain = '.'.join(parts[-2:])
subdomain = '.'.join(parts[:-2])
stats = domain_stats[base_domain]
stats["queries"] += 1
stats["total_qname_len"] += len(qname)
stats["subdomain_lengths"].append(len(subdomain))
stats["query_types"][qtype] += 1
stats["unique_subdomains"].add(subdomain)
# Noter les domaines pour les indicateurs de tunneling
suspicious = []
for domain, stats in domain_stats.items():
if stats["queries"] < 5:
continue
avg_subdomain_len = (sum(stats["subdomain_lengths"]) /
len(stats["subdomain_lengths"]))
unique_ratio = len(stats["unique_subdomains"]) / stats["queries"]
# Calculer l'entropie des sous-domaines
all_subdomains = ''.join(stats["unique_subdomains"])
sub_entropy = entropy(all_subdomains)
score = 0
reasons = []
if avg_subdomain_len > 30:
score += 30
reasons.append(f"Sous-domaines longs (moy {avg_subdomain_len:.0f} car.)")
if unique_ratio > 0.9:
score += 25
reasons.append(f"Haute unicité ({unique_ratio:.2%})")
if sub_entropy > 4.0:
score += 25
reasons.append(f"Haute entropie ({sub_entropy:.2f})")
if stats["query_types"].get(16, 0) > 10: # Enregistrements TXT
score += 20
reasons.append(f"Nombreuses requêtes TXT ({stats['query_types'][16]})")
if score >= 50:
suspicious.append({
"domain": domain,
"score": score,
"queries": stats["queries"],
"avg_subdomain_length": round(avg_subdomain_len, 1),
"unique_subdomains": len(stats["unique_subdomains"]),
"subdomain_entropy": round(sub_entropy, 2),
"reasons": reasons,
})
return sorted(suspicious, key=lambda x: -x["score"])
def analyze_icmp_tunneling(pcap_path):
"""Détecter le tunneling ICMP dans un PCAP."""
packets = rdpcap(pcap_path)
icmp_stats = defaultdict(lambda: {"count": 0, "payload_sizes": [], "payloads": []})
for pkt in packets:
if pkt.haslayer(ICMP) and pkt.haslayer(IP):
src = pkt[IP].src
dst = pkt[IP].dst
key = f"{src}->{dst}"
payload = bytes(pkt[ICMP].payload)
icmp_stats[key]["count"] += 1
icmp_stats[key]["payload_sizes"].append(len(payload))
if len(payload) > 64:
icmp_stats[key]["payloads"].append(payload[:100])
suspicious = []
for flow, stats in icmp_stats.items():
if stats["count"] < 5:
continue
avg_size = sum(stats["payload_sizes"]) / len(stats["payload_sizes"])
if avg_size > 64 or stats["count"] > 100:
suspicious.append({
"flow": flow,
"packets": stats["count"],
"avg_payload_size": round(avg_size, 1),
"reason": "Payloads ICMP volumineux/fréquents suggèrent un tunneling",
})
return suspicious
if __name__ == "__main__":
if len(sys.argv) < 2:
print(f"Usage: {sys.argv[0]} <fichier_pcap>")
sys.exit(1)
print("[+] Analyse du Tunneling DNS")
dns_results = analyze_dns_tunneling(sys.argv[1])
for r in dns_results:
print(f" {r['domain']} (score: {r['score']})")
for reason in r['reasons']:
print(f" - {reason}")
print("\n[+] Analyse du Tunneling ICMP")
icmp_results = analyze_icmp_tunneling(sys.argv[1])
for r in icmp_results:
print(f" {r['flow']}: {r['reason']}")
Critères de Validation
- Tunneling DNS détecté via l'entropie, la longueur des sous-domaines et l'analyse du volume de requêtes
- Canaux cachés ICMP identifiés par des anomalies de taille de payload
- Domaines de tunneling distingués du trafic CDN/cloud légitime
- Volume d'exfiltration de données estimé à partir du trafic capturé
- Modèles de communication C2 et intervalles de balisage extraits