Construction d'un Système de Suivi des Infrastructures Adverses
Vue d'ensemble
Le suivi des infrastructures adverses utilise les enregistrements DNS passifs, les logs de transparence des certificats, les données d'enregistrement WHOIS et l'enrichissement IP pour découvrir, cartographier et surveiller les réseaux de commande et contrôle (C2) des acteurs de menace. Les attaquants réutilisent fréquemment les hébergeurs, les bureaux d'enregistrement, les certificats SSL et les schémas de nommage à travers les campagnes, permettant aux analystes de pivoter à partir d'indicateurs connus pour découvrir de nouvelles infrastructures. Ce Skill couvre la construction d'un système automatisé de suivi qui identifie les relations entre infrastructures, détecte les domaines nouvellement enregistrés correspondant aux schémas des adversaires et maintient une carte continuellement mise à jour des réseaux des acteurs de menace.
Quand l'utiliser
- Lors du déploiement ou de la configuration de capacités de suivi des infrastructures adverses dans votre environnement
- Lors de l'établissement de contrôles de sécurité alignés sur les exigences de conformité
- Lors de la construction ou de l'amélioration de l'architecture de sécurité pour ce domaine
- Lors de la réalisation d'évaluations de sécurité nécessitant cette mise en œuvre
Prérequis
- Python 3.9+ avec les bibliothèques
requests,dnspython,python-whois,shodan,networkx - Clés API : SecurityTrails, PassiveTotal/RiskIQ, Shodan, VirusTotal
- Accès aux sources de données DNS passives
- Compréhension de l'infrastructure DNS, de l'hébergement et de l'enregistrement de domaines
- Base de données graphe (Neo4j) ou NetworkX pour la visualisation des relations
Concepts clés
DNS passif
Le DNS passif capture les données historiques de résolution DNS, enregistrant quels domaines ont été résolus vers quelles adresses IP et à quel moment. Contrairement aux requêtes DNS actives, le DNS passif préserve les relations historiques même après que les enregistrements aient changé, permettant aux analystes de suivre les changements d'infrastructure, d'identifier les schémas d'hébergement partagé et de découvrir des domaines connexes qui ont été résolus vers les mêmes adresses IP au fil du temps.
Pivotement d'infrastructure
Le pivotement identifie les infrastructures connexes en suivant les connexions : pivot IP (trouver tous les domaines sur une IP), pivot domaine (trouver toutes les IP auxquelles un domaine a été résolu), pivot WHOIS (trouver des domaines avec le même titulaire), pivot certificat (trouver des hôtes partageant des certificats SSL) et pivot NS/MX (trouver des domaines utilisant les mêmes serveurs de noms ou serveurs de messagerie).
Schémas d'infrastructure adverse
Les acteurs de menace présentent des schémas : bureaux d'enregistrement préférés (Namecheap, REG.RU, Tucows), hébergement préféré (fournisseurs d'hébergement à toute épreuve, services cloud), algorithmes de génération de domaines (DGA), schémas de nommage cohérents et réutilisation de certificats à travers les campagnes.
Workflow
Étape 1 : Découverte d'infrastructure DNS passive
import requests
import json
from collections import defaultdict
from datetime import datetime
class InfrastructureTracker:
def __init__(self, securitytrails_key=None, vt_key=None, shodan_key=None):
self.st_key = securitytrails_key
self.vt_key = vt_key
self.shodan_key = shodan_key
self.infrastructure_graph = defaultdict(lambda: {"nodes": set(), "edges": []})
def passive_dns_lookup(self, domain):
"""Interroge le DNS passif pour l'historique de résolution d'un domaine."""
headers = {"apikey": self.st_key}
url = f"https://api.securitytrails.com/v1/history/{domain}/dns/a"
resp = requests.get(url, headers=headers, timeout=30)
if resp.status_code == 200:
records = resp.json().get("records", [])
history = []
for record in records:
for value in record.get("values", []):
history.append({
"domain": domain,
"ip": value.get("ip", ""),
"first_seen": record.get("first_seen", ""),
"last_seen": record.get("last_seen", ""),
"type": record.get("type", "a"),
})
print(f"[+] DNS passif pour {domain} : {len(history)} enregistrements")
return history
return []
def reverse_ip_lookup(self, ip_address):
"""Trouve tous les domaines hébergés sur une adresse IP."""
headers = {"apikey": self.st_key}
url = f"https://api.securitytrails.com/v1/ips/nearby/{ip_address}"
resp = requests.get(url, headers=headers, timeout=30)
if resp.status_code == 200:
blocks = resp.json().get("blocks", [])
domains = []
for block in blocks:
for site in block.get("sites", []):
domains.append(site)
print(f"[+] IP inversée pour {ip_address} : {len(domains)} domaines")
return domains
return []
def whois_lookup(self, domain):
"""Obtient les données d'enregistrement WHOIS pour le pivotement."""
headers = {"apikey": self.st_key}
url = f"https://api.securitytrails.com/v1/domain/{domain}/whois"
resp = requests.get(url, headers=headers, timeout=30)
if resp.status_code == 200:
data = resp.json()
whois_data = {
"domain": domain,
"registrar": data.get("registrar", ""),
"registrant_org": data.get("registrant_org", ""),
"registrant_email": data.get("registrant_email", ""),
"name_servers": data.get("nameServers", []),
"created_date": data.get("createdDate", ""),
"updated_date": data.get("updatedDate", ""),
"expires_date": data.get("expiresDate", ""),
}
return whois_data
return {}
def pivot_from_seed(self, seed_indicator, indicator_type="domain", depth=2):
"""Pivote récursivement à partir d'un indicateur de départ pour découvrir l'infrastructure."""
discovered = {"domains": set(), "ips": set(), "relationships": []}
if indicator_type == "domain":
discovered["domains"].add(seed_indicator)
# Obtient les IP pour le domaine
pdns = self.passive_dns_lookup(seed_indicator)
for record in pdns:
ip = record["ip"]
discovered["ips"].add(ip)
discovered["relationships"].append({
"source": seed_indicator, "target": ip,
"type": "resolves_to",
"first_seen": record["first_seen"],
"last_seen": record["last_seen"],
})
if depth > 1:
# Recherche inversée sur les IP découvertes
reverse_domains = self.reverse_ip_lookup(ip)
for rd in reverse_domains[:20]:
discovered["domains"].add(rd)
discovered["relationships"].append({
"source": rd, "target": ip,
"type": "hosted_on",
})
elif indicator_type == "ip":
discovered["ips"].add(seed_indicator)
domains = self.reverse_ip_lookup(seed_indicator)
for domain in domains[:20]:
discovered["domains"].add(domain)
discovered["relationships"].append({
"source": domain, "target": seed_indicator,
"type": "hosted_on",
})
print(f"[+] Pivotement à partir de {seed_indicator} : "
f"{len(discovered['domains'])} domaines, "
f"{len(discovered['ips'])} IPs, "
f"{len(discovered['relationships'])} relations")
return discovered
tracker = InfrastructureTracker(
securitytrails_key="VOTRE_CLE_ST",
vt_key="VOTRE_CLE_VT",
)
Étape 2 : Construction du graphe d'infrastructure
import networkx as nx
class InfrastructureGraph:
def __init__(self):
self.graph = nx.Graph()
def add_discovery(self, discovery_data):
"""Ajoute l'infrastructure découverte au graphe."""
for domain in discovery_data["domains"]:
self.graph.add_node(domain, type="domain")
for ip in discovery_data["ips"]:
self.graph.add_node(ip, type="ip")
for rel in discovery_data["relationships"]:
self.graph.add_edge(
rel["source"], rel["target"],
relationship=rel["type"],
first_seen=rel.get("first_seen", ""),
last_seen=rel.get("last_seen", ""),
)
def find_clusters(self):
"""Identifie les clusters d'infrastructure."""
components = list(nx.connected_components(self.graph))
clusters = []
for component in components:
domains = [n for n in component if self.graph.nodes[n].get("type") == "domain"]
ips = [n for n in component if self.graph.nodes[n].get("type") == "ip"]
clusters.append({
"size": len(component),
"domains": sorted(domains),
"ips": sorted(ips),
"domain_count": len(domains),
"ip_count": len(ips),
})
clusters.sort(key=lambda x: x["size"], reverse=True)
print(f"[+] Clusters d'infrastructure : {len(clusters)}")
return clusters
def find_hub_nodes(self, top_n=10):
"""Trouve les nœuds à haute centralité (infrastructure partagée)."""
centrality = nx.degree_centrality(self.graph)
top_nodes = sorted(centrality.items(), key=lambda x: x[1], reverse=True)[:top_n]
hubs = []
for node, score in top_nodes:
hubs.append({
"node": node,
"type": self.graph.nodes[node].get("type", "unknown"),
"centrality": round(score, 4),
"connections": self.graph.degree(node),
})
return hubs
def export_graph(self, output_file="infrastructure_graph.json"):
data = nx.node_link_data(self.graph)
with open(output_file, "w") as f:
json.dump(data, f, indent=2)
print(f"[+] Graphe exporté : {self.graph.number_of_nodes()} nœuds, "
f"{self.graph.number_of_edges()} arêtes")
infra_graph = InfrastructureGraph()
discovery = tracker.pivot_from_seed("evil-domain.com", depth=2)
infra_graph.add_discovery(discovery)
clusters = infra_graph.find_clusters()
hubs = infra_graph.find_hub_nodes()
infra_graph.export_graph()
Étape 3 : Surveillance des nouvelles infrastructures
import time
class InfrastructureMonitor:
def __init__(self, tracker, known_indicators):
self.tracker = tracker
self.known = set(known_indicators)
self.alerts = []
def check_new_registrations(self, patterns):
"""Vérifie les domaines nouvellement enregistrés correspondant aux schémas adverses."""
import re
new_domains = []
for pattern in patterns:
# Interroge SecurityTrails pour les nouveaux domaines correspondant au schéma
headers = {"apikey": self.tracker.st_key}
url = "https://api.securitytrails.com/v1/domains/list"
params = {"include_ips": "true", "page": 1}
body = {"filter": {"keyword": pattern}}
resp = requests.post(url, headers=headers, json=body, timeout=30)
if resp.status_code == 200:
records = resp.json().get("records", [])
for record in records:
domain = record.get("hostname", "")
if domain not in self.known:
new_domains.append({
"domain": domain,
"pattern_matched": pattern,
"first_seen": datetime.now().isoformat(),
})
self.known.add(domain)
if new_domains:
print(f"[ALERTE] {len(new_domains)} nouveaux domaines correspondant aux schémas")
self.alerts.extend(new_domains)
return new_domains
def generate_infrastructure_report(self, clusters, hubs):
report = f"""# Rapport de suivi des infrastructures adverses
Généré le : {datetime.now().isoformat()}
## Résumé
- Clusters d'infrastructure identifiés : {len(clusters)}
- Total de domaines suivis : {sum(c['domain_count'] for c in clusters)}
- Total d'IP suivies : {sum(c['ip_count'] for c in clusters)}
- Nouveaux domaines détectés : {len(self.alerts)}
## Principaux hubs d'infrastructure
| Nœud | Type | Connexions | Centralité |
|------|------|-------------|------------|
"""
for hub in hubs[:10]:
report += (f"| {hub['node']} | {hub['type']} "
f"| {hub['connections']} | {hub['centrality']} |\n")
report += "\n## Clusters d'infrastructure\n"
for i, cluster in enumerate(clusters[:5], 1):
report += f"\n### Cluster {i} ({cluster['size']} nœuds)\n"
report += f"- Domaines : {', '.join(cluster['domains'][:5])}\n"
report += f"- IPs : {', '.join(cluster['ips'][:5])}\n"
with open("infrastructure_report.md", "w") as f:
f.write(report)
print("[+] Rapport d'infrastructure enregistré")
monitor = InfrastructureMonitor(tracker, known_indicators=set())
Critères de validation
- Les requêtes DNS passives renvoient des données de résolution historiques
- Les recherches IP inversées découvrent les domaines co-hébergés
- Le pivotement d'infrastructure s'étend à partir d'indicateurs de départ
- L'analyse du graphe identifie les clusters et les nœuds hubs
- Les nouvelles infrastructures sont détectées via la surveillance de schémas
- Des rapports sont générés avec des recommandations actionnables