$npx -y skills add killvxk/cybersecurity-skills-zh --skill analyzing-network-covert-channels-in-malware检测和分析恶意软件使用的隐蔽通信信道,包括 DNS 隧道、ICMP 数据泄露、HTTP 隐写术和协议滥用,用于 C2 通信和数据泄露。
| 1 | # 分析恶意软件中的网络隐蔽信道 |
| 2 | |
| 3 | ## 概述 |
| 4 | |
| 5 | 恶意软件使用隐蔽信道将 C2 通信和数据泄露伪装成看似合法的网络流量。DNS 隧道将数据编码在 DNS 查询和响应中(iodine、dnscat2 等工具和 FrameworkPOS 等恶意软件家族均使用此技术)。ICMP 隧道将数据隐藏在回显请求/响应载荷中(icmpsh、ptunnel)。HTTP 隐蔽信道将 C2 数据嵌入头部、Cookie 或隐写图像中。协议滥用利用允许的协议绕过防火墙。现代基于机器学习的方法对 DNS 隧道检测可达 99% 以上的召回率,但低吞吐量泄露仍然具有挑战性。Palo Alto Unit42 在 2024 年跟踪了三个主要的 DNS 隧道活动(TrkCdn、SecShow、Savvy Seahorse),显示了该技术的持续普遍性。 |
| 6 | |
| 7 | ## 前置条件 |
| 8 | |
| 9 | - Python 3.9+,安装 `scapy`、`dpkt`、`dnslib` |
| 10 | - Wireshark/tshark,用于 PCAP 分析 |
| 11 | - Zeek(原 Bro),用于网络监控 |
| 12 | - DNS 查询日志基础设施 |
| 13 | - 对 DNS、ICMP、HTTP 协议的数据包级理解 |
| 14 | |
| 15 | ## 操作步骤 |
| 16 | |
| 17 | ### 步骤 1:DNS 隧道检测 |
| 18 | |
| 19 | ```python |
| 20 | #!/usr/bin/env python3 |
| 21 | """检测网络流量中的 DNS 隧道和隐蔽信道。""" |
| 22 | import sys |
| 23 | import json |
| 24 | import math |
| 25 | from collections import Counter, defaultdict |
| 26 | |
| 27 | try: |
| 28 | from scapy.all import rdpcap, DNS, DNSQR, DNSRR, IP, ICMP |
| 29 | except ImportError: |
| 30 | print("pip install scapy") |
| 31 | sys.exit(1) |
| 32 | |
| 33 | |
| 34 | def entropy(data): |
| 35 | if not data: |
| 36 | return 0 |
| 37 | freq = Counter(data) |
| 38 | length = len(data) |
| 39 | return -sum((c/length) * math.log2(c/length) for c in freq.values()) |
| 40 | |
| 41 | |
| 42 | def analyze_dns_tunneling(pcap_path): |
| 43 | """检测 PCAP 中的 DNS 隧道指标。""" |
| 44 | packets = rdpcap(pcap_path) |
| 45 | domain_stats = defaultdict(lambda: { |
| 46 | "queries": 0, "total_qname_len": 0, "subdomain_lengths": [], |
| 47 | "query_types": Counter(), "unique_subdomains": set(), |
| 48 | }) |
| 49 | |
| 50 | for pkt in packets: |
| 51 | if pkt.haslayer(DNS) and pkt.haslayer(DNSQR): |
| 52 | qname = pkt[DNSQR].qname.decode('utf-8', errors='replace').rstrip('.') |
| 53 | qtype = pkt[DNSQR].qtype |
| 54 | |
| 55 | parts = qname.split('.') |
| 56 | if len(parts) >= 3: |
| 57 | base_domain = '.'.join(parts[-2:]) |
| 58 | subdomain = '.'.join(parts[:-2]) |
| 59 | |
| 60 | stats = domain_stats[base_domain] |
| 61 | stats["queries"] += 1 |
| 62 | stats["total_qname_len"] += len(qname) |
| 63 | stats["subdomain_lengths"].append(len(subdomain)) |
| 64 | stats["query_types"][qtype] += 1 |
| 65 | stats["unique_subdomains"].add(subdomain) |
| 66 | |
| 67 | # 对域名进行隧道指标评分 |
| 68 | suspicious = [] |
| 69 | for domain, stats in domain_stats.items(): |
| 70 | if stats["queries"] < 5: |
| 71 | continue |
| 72 | |
| 73 | avg_subdomain_len = (sum(stats["subdomain_lengths"]) / |
| 74 | len(stats["subdomain_lengths"])) |
| 75 | unique_ratio = len(stats["unique_subdomains"]) / stats["queries"] |
| 76 | |
| 77 | # 计算子域名熵 |
| 78 | all_subdomains = ''.join(stats["unique_subdomains"]) |
| 79 | sub_entropy = entropy(all_subdomains) |
| 80 | |
| 81 | score = 0 |
| 82 | reasons = [] |
| 83 | |
| 84 | if avg_subdomain_len > 30: |
| 85 | score += 30 |
| 86 | reasons.append(f"子域名过长(平均 {avg_subdomain_len:.0f} 字符)") |
| 87 | if unique_ratio > 0.9: |
| 88 | score += 25 |
| 89 | reasons.append(f"唯一性高({unique_ratio:.2%})") |
| 90 | if sub_entropy > 4.0: |
| 91 | score += 25 |
| 92 | reasons.append(f"熵值高({sub_entropy:.2f})") |
| 93 | if stats["query_types"].get(16, 0) > 10: # TXT 记录 |
| 94 | score += 20 |
| 95 | reasons.append(f"大量 TXT 查询({stats['query_types'][16]} 次)") |
| 96 | |
| 97 | if score >= 50: |
| 98 | suspicious.append({ |
| 99 | "domain": domain, |
| 100 | "score": score, |
| 101 | "queries": stats["queries"], |
| 102 | "avg_subdomain_length": round(avg_subdomain_len, 1), |
| 103 | "unique_subdomains": len(stats["unique_subdomains"]), |
| 104 | "subdomain_entropy": round(sub_entropy, 2), |
| 105 | "reasons": reasons, |
| 106 | }) |
| 107 | |
| 108 | return sorted(suspicious, key=lambda x: -x["score"]) |
| 109 | |
| 110 | |
| 111 | def analyze_icmp_tunneling(pcap_path): |
| 112 | """检测 PCAP 中的 ICMP 隧道。""" |
| 113 | packets = rdpcap(pcap_path) |
| 114 | icmp_stats = defaultdict(lambda: {"count": 0, "payload_sizes": [], "payloads": []}) |
| 115 | |
| 116 | for pkt in packets: |
| 117 | if pkt.haslayer(ICMP) and pkt.haslayer(IP): |
| 118 | src = pkt[IP].src |
| 119 | dst = pkt[IP].dst |
| 120 | key = f"{src}->{dst}" |
| 121 | |
| 122 | payload = bytes(pkt[ICMP].payload) |
| 123 | icmp_stats[key]["count"] += 1 |
| 124 | icmp_stats[key]["payload_sizes"].append(len(payload)) |
| 125 | if len(payload) > 64: |
| 126 | icmp_stats[key]["payloads"].append(payload[:100]) |
| 127 | |
| 128 | suspicious = [] |
| 129 | for flow, stats in icmp_stats.items(): |
| 130 | if stats["count"] < 5: |
| 131 | continue |
| 132 | avg_size = sum(stats["payload_sizes"]) / len(stats["payload_sizes"]) |
| 133 | if avg_size > 64 or stats["count"] > 100: |
| 134 | suspicious.append({ |
| 135 | "flow": flow, |
| 136 | "packets": stats["count"], |
| 137 | "avg_payload_size": round(avg_size, 1), |
| 138 | "reason": "大型/频繁的 ICMP 载荷表明存在隧道", |
| 139 | }) |
| 140 | |
| 141 | return suspicious |
| 142 | |
| 143 | |
| 144 | if __name__ == "__main__": |
| 145 | if len(sys.argv) < 2: |
| 146 | print(f"用法:{sys.argv |