Bau eines Systems zur Verfolgung von Angreifer-Infrastruktur
Übersicht
Die Verfolgung von Angreifer-Infrastruktur nutzt passive DNS-Daten, Certificate-Transparency-Logs, WHOIS-Registrierungsdaten und IP-Anreicherung, um Command-and-Control-Netzwerke (C2) von Bedrohungsakteuren zu entdecken, zu kartieren und zu überwachen. Angreifer verwenden häufig Hosting-Anbieter, Registrar, SSL-Zertifikate und Namensmuster über mehrere Kampagnen hinweg wieder. Dies ermöglicht Analysten, von bekannten Indikatoren zu pivotieren und neue Infrastruktur zu entdecken. Diese Fähigkeit umfasst den Bau eines automatisierten Verfolgungssystems, das Infrastrukturbeziehungen identifiziert, neu registrierte Domains erkennt, die mit Angreifermustern übereinstimmen, und eine kontinuierlich aktualisierte Karte von Netzwerken der Bedrohungsakteure pflegt.
Wann zu verwenden
- Beim Bereitstellen oder Konfigurieren von Funktionen zur Verfolgung von Angreifer-Infrastruktur in Ihrer Umgebung
- Bei der Einrichtung von Sicherheitskontrollen nach Compliance-Anforderungen
- Beim Aufbau oder der Verbesserung der Sicherheitsarchitektur für diesen Bereich
- Bei der Durchführung von Sicherheitsbewertungen, die diese Implementierung erfordern
Voraussetzungen
- Python 3.9+ mit den Bibliotheken
requests,dnspython,python-whois,shodan,networkx - API-Schlüssel: SecurityTrails, PassiveTotal/RiskIQ, Shodan, VirusTotal
- Zugang zu passiven DNS-Datenquellen
- Verständnis von DNS-Infrastruktur, Hosting und Domain-Registrierung
- Graphdatenbank (Neo4j) oder NetworkX für die Beziehungsvisualisierung
Schlüsselkonzepte
Passives DNS
Passives DNS erfasst historische DNS-Auflösungsdaten und zeichnet auf, welche Domains zu welchen IPs aufgelöst wurden und zu welchem Zeitpunkt. Im Gegensatz zu aktiven DNS-Abfragen bleiben bei passivem DNS historische Beziehungen erhalten, auch wenn sich Einträge ändern. Dies ermöglicht Analysten, Infrastrukturänderungen zu verfolgen, gemeinsame Hosting-Muster zu identifizieren und verwandte Domains zu entdecken, die im Laufe der Zeit auf dieselben IP-Adressen aufgelöst wurden.
Infrastruktur-Pivotierung
Pivotierung identifiziert verwandte Infrastruktur durch Verfolgen von Verbindungen: IP-Pivot (alle Domains auf einer IP finden), Domain-Pivot (alle IPs finden, zu denen eine Domain aufgelöst wurde), WHOIS-Pivot (Domains mit demselben Registranten finden), Zertifikats-Pivot (Hosts finden, die SSL-Zertifikate teilen) und NS/MX-Pivot (Domains finden, die dieselben Nameserver oder Mailserver verwenden).
Muster von Angreifer-Infrastruktur
Bedrohungsakteure zeigen Muster: bevorzugte Registrar (Namecheap, REG.RU, Tucows), bevorzugtes Hosting (bulletproof Hosting-Anbieter, Cloud-Dienste), Domain-Generierungsalgorithmen (DGA), konsistente Namensmuster und Wiederverwendung von Zertifikaten über Kampagnen hinweg.
Arbeitsablauf
Schritt 1: Passive DNS-Infrastrukturentdeckung
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):
"""Passive DNS für den Auflösungsverlauf einer Domain abfragen."""
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"[+] Passives DNS für {domain}: {len(history)} Einträge")
return history
return []
def reverse_ip_lookup(self, ip_address):
"""Alle Domains finden, die auf einer IP-Adresse gehostet werden."""
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"[+] Reverse IP für {ip_address}: {len(domains)} Domains")
return domains
return []
def whois_lookup(self, domain):
"""WHOIS-Registrierungsdaten für Pivotierung abrufen."""
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):
"""Rekursiv von einem Startindikator pivotieren, um Infrastruktur zu entdecken."""
discovered = {"domains": set(), "ips": set(), "relationships": []}
if indicator_type == "domain":
discovered["domains"].add(seed_indicator)
# IPs für die Domain abrufen
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:
# Reverse-Lookup auf entdeckten IPs
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"[+] Pivot von {seed_indicator}: "
f"{len(discovered['domains'])} Domains, "
f"{len(discovered['ips'])} IPs, "
f"{len(discovered['relationships'])} Beziehungen")
return discovered
tracker = InfrastructureTracker(
securitytrails_key="DEIN_ST_KEY",
vt_key="DEIN_VT_KEY",
)
Schritt 2: Infrastrukturgraphen aufbauen
import networkx as nx
class InfrastructureGraph:
def __init__(self):
self.graph = nx.Graph()
def add_discovery(self, discovery_data):
"""Entdeckte Infrastruktur zum Graphen hinzufügen."""
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):
"""Infrastrukturcluster identifizieren."""
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"[+] Infrastrukturcluster: {len(clusters)}")
return clusters
def find_hub_nodes(self, top_n=10):
"""Knoten mit hoher Zentralität (gemeinsam genutzte Infrastruktur) finden."""
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"[+] Graph exportiert: {self.graph.number_of_nodes()} Knoten, "
f"{self.graph.number_of_edges()} Kanten")
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()
Schritt 3: Neue Infrastruktur überwachen
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):
"""Neu registrierte Domains prüfen, die mit Angreifermustern übereinstimmen."""
import re
new_domains = []
for pattern in patterns:
# SecurityTrails nach neuen Domains abfragen, die dem Muster entsprechen
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"[ALARM] {len(new_domains)} neue Domains entsprechen Mustern")
self.alerts.extend(new_domains)
return new_domains
def generate_infrastructure_report(self, clusters, hubs):
report = f"""# Bericht zur Verfolgung von Angreifer-Infrastruktur
Erstellt: {datetime.now().isoformat()}
## Zusammenfassung
- Identifizierte Infrastrukturcluster: {len(clusters)}
- Überwachte Domains gesamt: {sum(c['domain_count'] for c in clusters)}
- Überwachte IPs gesamt: {sum(c['ip_count'] for c in clusters)}
- Neue erkannte Domains: {len(self.alerts)}
## Top-Infrastruktur-Knoten
| Knoten | Typ | Verbindungen | Zentralität |
|--------|-----|--------------|-------------|
"""
for hub in hubs[:10]:
report += (f"| {hub['node']} | {hub['type']} "
f"| {hub['connections']} | {hub['centrality']} |\n")
report += "\n## Infrastrukturcluster\n"
for i, cluster in enumerate(clusters[:5], 1):
report += f"\n### Cluster {i} ({cluster['size']} Knoten)\n"
report += f"- Domains: {', '.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("[+] Infrastrukturbericht gespeichert")
monitor = InfrastructureMonitor(tracker, known_indicators=set())
Validierungskriterien
- Passive DNS-Abfragen geben historische Auflösungsdaten zurück
- Reverse IP-Lookups entdecken gemeinsam gehostete Domains
- Infrastruktur-Pivotierung erweitert sich ausgehend von Startindikatoren
- Graphenanalyse identifiziert Cluster und Knoten mit hoher Zentralität
- Neue Infrastruktur wird durch Musterüberwachung erkannt
- Berichte mit umsetzbaren Empfehlungen werden erstellt