| name | analyzing-network-covert-channels-in-malware |
| description | 检测和分析恶意软件使用的隐蔽通信信道，包括 DNS 隧道、ICMP 数据泄露、HTTP 隐写术和协议滥用，用于 C2 通信和数据泄露。 |
| domain | cybersecurity |
| subdomain | malware-analysis |
| tags | ["covert-channels","dns-tunneling","icmp-exfiltration","malware-analysis","network-forensics","c2-detection","data-exfiltration"] |
| version | 1.0 |
| author | mahipal |
| license | Apache-2.0 |
分析恶意软件中的网络隐蔽信道
概述
恶意软件使用隐蔽信道将 C2 通信和数据泄露伪装成看似合法的网络流量。DNS 隧道将数据编码在 DNS 查询和响应中（iodine、dnscat2 等工具和 FrameworkPOS 等恶意软件家族均使用此技术）。ICMP 隧道将数据隐藏在回显请求/响应载荷中（icmpsh、ptunnel）。HTTP 隐蔽信道将 C2 数据嵌入头部、Cookie 或隐写图像中。协议滥用利用允许的协议绕过防火墙。现代基于机器学习的方法对 DNS 隧道检测可达 99% 以上的召回率，但低吞吐量泄露仍然具有挑战性。Palo Alto Unit42 在 2024 年跟踪了三个主要的 DNS 隧道活动（TrkCdn、SecShow、Savvy Seahorse），显示了该技术的持续普遍性。
前置条件
- Python 3.9+，安装
scapy、dpkt、dnslib
- Wireshark/tshark，用于 PCAP 分析
- Zeek（原 Bro），用于网络监控
- DNS 查询日志基础设施
- 对 DNS、ICMP、HTTP 协议的数据包级理解
操作步骤
步骤 1：DNS 隧道检测
"""检测网络流量中的 DNS 隧道和隐蔽信道。"""
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):
"""检测 PCAP 中的 DNS 隧道指标。"""
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()
(parts) >= :
base_domain = .join(parts[-:])
subdomain = .join(parts[:-])
stats = domain_stats[base_domain]
stats[] +=
stats[] += (qname)
stats[].append((subdomain))
stats[][qtype] +=
stats[].add(subdomain)
suspicious = []
domain, stats domain_stats.items():
stats[] < :
avg_subdomain_len = ((stats[]) /
(stats[]))
unique_ratio = (stats[]) / stats[]
all_subdomains = .join(stats[])
sub_entropy = entropy(all_subdomains)
score =
reasons = []
avg_subdomain_len > :
score +=
reasons.append()
unique_ratio > :
score +=
reasons.append()
sub_entropy > :
score +=
reasons.append()
stats[].get(, ) > :
score +=
reasons.append()
score >= :
suspicious.append({
: domain,
: score,
: stats[],
: (avg_subdomain_len, ),
: (stats[]),
: (sub_entropy, ),
: reasons,
})
(suspicious, key= x: -x[])
():
packets = rdpcap(pcap_path)
icmp_stats = defaultdict(: {: , : [], : []})
pkt packets:
pkt.haslayer(ICMP) pkt.haslayer(IP):
src = pkt[IP].src
dst = pkt[IP].dst
key =
payload = (pkt[ICMP].payload)
icmp_stats[key][] +=
icmp_stats[key][].append((payload))
(payload) > :
icmp_stats[key][].append(payload[:])
suspicious = []
flow, stats icmp_stats.items():
stats[] < :
avg_size = (stats[]) / (stats[])
avg_size > stats[] > :
suspicious.append({
: flow,
: stats[],
: (avg_size, ),
: ,
})
suspicious
__name__ == :
(sys.argv) < :
()
sys.exit()
()
dns_results = analyze_dns_tunneling(sys.argv[])
r dns_results:
()
reason r[]:
()
()
icmp_results = analyze_icmp_tunneling(sys.argv[])
r icmp_results:
()
验证标准
- 通过熵值、子域名长度和查询量分析检测 DNS 隧道
- 通过载荷大小异常识别 ICMP 隐蔽信道
- 区分隧道域名和合法的 CDN/云流量
- 从捕获的流量中估算数据泄露量
- 提取 C2 通信模式和信标间隔
参考资料
