# Analyzing Network Covert Channels In Malware

> 检测和分析恶意软件使用的隐蔽通信信道，包括 DNS 隧道、ICMP 数据泄露、HTTP 隐写术和协议滥用，用于 C2 通信和数据泄露。

- Skill: `killvxk/analyzing-network-covert-channels-in-malware` (Agent Skill, multi-file: 7 files)
- Install (CLI): `npx skillmds@latest add killvxk/analyzing-network-covert-channels-in-malware`
- Raw SKILL.md: https://api.skillmd.com/api/skills/killvxk/analyzing-network-covert-channels-in-malware/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- License: Apache-2.0
- Author: killvxk (https://skillmd.com/u/killvxk)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/killvxk/analyzing-network-covert-channels-in-malware

---

# 分析恶意软件中的网络隐蔽信道

## 概述

恶意软件使用隐蔽信道将 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 隧道检测

```python
#!/usr/bin/env python3
"""检测网络流量中的 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('.')
            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)

    # 对域名进行隧道指标评分
    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"]

        # 计算子域名熵
        all_subdomains = ''.join(stats["unique_subdomains"])
        sub_entropy = entropy(all_subdomains)

        score = 0
        reasons = []

        if avg_subdomain_len > 30:
            score += 30
            reasons.append(f"子域名过长（平均 {avg_subdomain_len:.0f} 字符）")
        if unique_ratio > 0.9:
            score += 25
            reasons.append(f"唯一性高（{unique_ratio:.2%}）")
        if sub_entropy > 4.0:
            score += 25
            reasons.append(f"熵值高（{sub_entropy:.2f}）")
        if stats["query_types"].get(16, 0) > 10:  # TXT 记录
            score += 20
            reasons.append(f"大量 TXT 查询（{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):
    """检测 PCAP 中的 ICMP 隧道。"""
    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": "大型/频繁的 ICMP 载荷表明存在隧道",
            })

    return suspicious


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"用法：{sys.argv[0]} <pcap_file>")
        sys.exit(1)

    print("[+] DNS 隧道分析")
    dns_results = analyze_dns_tunneling(sys.argv[1])
    for r in dns_results:
        print(f"  {r['domain']}（评分：{r['score']}）")
        for reason in r['reasons']:
            print(f"    - {reason}")

    print("\n[+] ICMP 隧道分析")
    icmp_results = analyze_icmp_tunneling(sys.argv[1])
    for r in icmp_results:
        print(f"  {r['flow']}：{r['reason']}")
```

## 验证标准

- 通过熵值、子域名长度和查询量分析检测 DNS 隧道
- 通过载荷大小异常识别 ICMP 隐蔽信道
- 区分隧道域名和合法的 CDN/云流量
- 从捕获的流量中估算数据泄露量
- 提取 C2 通信模式和信标间隔

## 参考资料

- [Palo Alto Unit42 - DNS 隧道活动](https://unit42.paloaltonetworks.com/three-dns-tunneling-campaigns/)
- [Elastic - 检测隐蔽数据泄露](https://www.elastic.co/blog/elastic-security-detecting-covert-data-exfiltration)
- [Vectra AI - ICMP 隧道检测](https://www.vectra.ai/detections/icmp-tunnel)
- [MITRE ATT&CK T1071 - 应用层协议](https://attack.mitre.org/techniques/T1071/)