Was macht Analyse von Netzwerk-Covert-Kanälen in Malware?
Übersicht
Malware nutzt Covert-Kanäle, um C2-Kommunikation und Datenaustritt in vermeintlich legitimen Netzwerkverkehr zu tarnen. DNS-Tunneling kodiert Daten in DNS-Abfragen und -Antworten (genutzt von Tools wie iodine, dnscat2 und Malware-Familien wie FrameworkPOS). ICMP-Tunneling versteckt Daten in Echo-Request/Reply-Payloads (icmpsh, ptunnel). HTTP-Covert-Kanäle betten C2-Daten in Header, Cookies oder steganografische Bilder ein. Protokollmissbrauch nutzt erlaubte Protokolle, um Firewalls zu umgehen. Die Erkennung von DNS-Tunneling erreicht mit modernen ML-basierten Ansätzen eine Trefferquote von über 99 %, wobei geringvolumige Datenaustritte weiterhin eine Herausforderung darstellen. Palo Alto Unit42 verfolgte bis 2024 drei große DNS-Tunneling-Kampagnen (TrkCdn, SecShow, Savvy Seahorse), was die anhaltende Verbreitung dieser Technik zeigt.
Verwendung
- Bei der Untersuchung von Sicherheitsvorfällen, die die Analyse von Netzwerk-Covert-Kanälen in Malware erfordern
- Bei der Erstellung von Erkennungsregeln oder Bedrohungsjagd-Abfragen für diesen Bereich
- Wenn SOC-Analysten strukturierte Verfahren für diese Art von Analyse benötigen
- Bei der Validierung der Sicherheitsüberwachungsabdeckung für verwandte Angriffstechniken
Voraussetzungen
- Python 3.9+ mit
scapy,dpkt,dnslib - Wireshark/tshark für PCAP-Analyse
- Zeek (ehemals Bro) für Netzwerküberwachung
- DNS-Abfrageprotokollierungsinfrastruktur
- Verständnis von DNS-, ICMP- und HTTP-Protokollen auf Paketebene
Arbeitsablauf
Schritt 1: DNS-Tunneling-Erkennung
#!/usr/bin/env python3
"""Erkenne DNS-Tunneling und Covert-Kanäle in Netzwerkverkehr."""
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):
"""Erkenne DNS-Tunneling-Indikatoren in 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)
# Domänen auf Tunneling-Indikatoren bewerten
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"]
# Subdomain-Entropie berechnen
all_subdomains = ''.join(stats["unique_subdomains"])
sub_entropy = entropy(all_subdomains)
score = 0
reasons = []
if avg_subdomain_len > 30:
score += 30
reasons.append(f"Lange Subdomains (avg {avg_subdomain_len:.0f} Zeichen)")
if unique_ratio > 0.9:
score += 25
reasons.append(f"Hohe Einzigartigkeit ({unique_ratio:.2%})")
if sub_entropy > 4.0:
score += 25
reasons.append(f"Hohe Entropie ({sub_entropy:.2f})")
if stats["query_types"].get(16, 0) > 10: # TXT-Einträge
score += 20
reasons.append(f"Viele TXT-Abfragen ({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):
"""Erkenne ICMP-Tunneling in 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": "Große/häufige ICMP-Payloads deuten auf Tunneling hin",
})
return suspicious
if __name__ == "__main__":
if len(sys.argv) < 2:
print(f"Verwendung: {sys.argv[0]} <pcap_datei>")
sys.exit(1)
print("[+] DNS-Tunneling-Analyse")
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[+] ICMP-Tunneling-Analyse")
icmp_results = analyze_icmp_tunneling(sys.argv[1])
for r in icmp_results:
print(f" {r['flow']}: {r['reason']}")
Validierungskriterien
- DNS-Tunneling erkannt über Entropie, Subdomain-Länge und Abfragevolumen-Analyse
- ICMP-Covert-Kanäle identifiziert durch Auffälligkeiten bei der Payload-Größe
- Tunneling-Domänen von legitimen CDN/Cloud-Verkehr unterschieden
- Datenaustritt-Volumen aus erfasstem Verkehr geschätzt
- C2-Kommunikationsmuster und Beaconing-Intervalle extrahiert