简介这份资源包面向网络安全初学者与进阶研究者聚焦利用DNS流量分析识别僵尸网络异常行为提供从数据采集、特征提取到模型训练与评估的完整实践方案。包内共27个文件以15个Python脚本为核心覆盖DNS解析、流量抓取、特征工程与检测逻辑另有6个pyc编译文件、2个txt与2个md说明文档、1个readme及1个ipynb笔记本压缩包约3.06MB结构清晰便于按模块查阅。内容涉及查询频率、查询类型、源IP分布、响应延迟等特征并对比支持向量机、随机森林、K-means、Isolation Forest等机器学习方法以及CNN、RNN、Transformer在时序流量上的应用思路。已有313人学习适合希望掌握异常检测流程、复现实验环境并理解DNS安全分析要点的读者参考。1. DNS 流量里的僵尸网络为什么它是最隐蔽的检测入口DNS 是互联网里最不起眼、却最不可能被关掉的协议。一台主机可以没有 HTTP 流量、可以没有 SSH 登录但只要它想跟外界保持联系几乎必然要发 DNS 查询。僵尸网络Botnet的操控者深谙这一点他们把 C2命令与控制指令塞进 DNS 查询名里用 TXT、A、AAAA 记录回传数据甚至用 DNS 隧道把整条 TCP 通道伪装成域名解析。结果就是——防火墙看着 53 端口一切正常流量分析工具却抓不到任何可疑的 HTTP 特征。这个标题要解决的核心问题就是在只有 DNS 流量的前提下把受控主机和恶意域名从海量正常解析里揪出来。它适合两类人一是做企业内网安全的工程师手上有 DNS 日志或镜像流量但不知道怎么下手二是做流量分析 CTF 或安全研究的同学想从 DNS 这个切面建立一套可复现的检测流程。我下面讲的方案不依赖任何商业威胁情报只用公开数据和开源工具从特征提取到规则落地一步步走通。2. 僵尸网络在 DNS 层留下的四类指纹2.1 DGA 域名为什么随机字符串是第一个抓手DGADomain Generation Algorithm是僵尸网络最经典的存活手段。恶意程序内置一个算法每天生成成百上千个候选域名逐个尝试解析其中只有少数几个会被攻击者提前注册并解析到 C2。对检测方来说这带来一个非常稳定的统计特征大量解析失败的随机域名。正常用户也会打错域名但比例极低且错误域名往往接近真实域名编辑距离小。DGA 域名则不同长度集中在 820 字符字符分布接近均匀随机元音辅音比例异常n-gram 得分低。你可以用几个轻量指标先做粗筛指标正常域名典型值DGA 域名典型值域名长度6151025数字占比 10%15%40%最长连续辅音2346唯一字符数 / 长度0.50.70.70.95解析失败率低高这些指标不需要机器学习模型用 Python 十几行就能算。关键是别只看单条查询要看一个时间窗口内的聚合。单次随机域名可能是用户手滑但一台主机在 5 分钟内发起 50 次以上失败解析且域名熵值普遍偏高基本可以判定异常。2.2 DNS 隧道TXT 记录和超长子域名DNS 隧道是把数据编码进域名或 TXT 记录里实现双向通信。检测特征比 DGA 更明确子域名部分异常长单标签超过 50 字符很常见TXT 记录查询占比异常升高同一父域名下出现大量不同子域名查询类型集中为 TXT、NULL、CNAME 等非 A 记录请求/响应包大小分布与正常 DNS 明显不同我一般会先统计每个父域名的子域名基数和平均标签长度。正常 CDN 域名子域名也多但标签长度短且字符集规整隧道域名的子域名往往像 Base64 编码长度整齐、字符集包含大小写字母加数字。2.3 Fast Flux 与域名复用Fast Flux 指同一个域名在短时间内解析到大量不同 IP且 TTL 极短。僵尸网络用这种方式隐藏真实 C2 主机。检测思路是对每个域名统计单位时间内的唯一 A 记录数量和TTL 分布。正常业务域名可能用 CDN但 TTL 不会低到 30 秒以下还频繁换 IP。如果某个域名 10 分钟内解析出 20 个以上不同 IP且 TTL 普遍小于 60 秒就值得拉出来单独看。2.4 心跳与周期性僵尸网络需要定期回连 C2 领取指令这会在 DNS 查询时间序列上留下周期性。正常用户的 DNS 查询跟作息相关白天多、夜间少但不会精确到每 60 秒一次。你可以对每台主机的查询时间戳做差分然后看差分序列的方差。方差极小、间隔稳定的就是心跳特征。3. 用 Python 把 DNS 日志跑成检测流水线3.1 数据准备从 pcap 到结构化记录假设你手上是 pcap 或者已经导出的 DNS 日志。我一般用dnspython加scapy做解析输出成一行一条的 JSON方便后续聚合。先装依赖pip install scapy dnspython pandas numpy解析脚本核心逻辑如下from scapy.all import rdpcap, DNS, DNSQR, DNSRR import json def parse_dns_pcap(pcap_path, out_path): packets rdpcap(pcap_path) records [] for pkt in packets: if pkt.haslayer(DNS): dns pkt[DNS] # 只取查询和响应忽略其他 if dns.qr 0 and dns.qd is not None: qname dns.qd.qname.decode(errorsignore).rstrip(.) qtype dns.qd.qtype records.append({ ts: float(pkt.time), src: pkt[IP].src if pkt.haslayer(IP) else , qname: qname, qtype: qtype, rcode: dns.rcode, ancount: dns.ancount, }) with open(out_path, w) as f: for r in records: f.write(json.dumps(r) \n) parse_dns_pcap(dns.pcap, dns_records.jsonl)这段代码只提取查询报文字段包括时间戳、源 IP、查询名、查询类型、响应码和答案数量。rcode为 3 表示 NXDOMAIN是 DGA 检测的关键字段。ancount为 0 且rcode为 0 的情况也要留意可能是空应答。参数上rdpcap对大文件吃内存超过 500MB 的 pcap 建议先用tcpdump切分tcpdump -r big.pcap -w part_%03d.pcap -C 1003.2 特征计算熵、n-gram 和失败率拿到结构化记录后按「源 IP 时间窗口」聚合。下面这个函数计算单个域名的几个核心特征import math from collections import Counter def shannon_entropy(s): if not s: return 0.0 cnt Counter(s) n len(s) return -sum((c / n) * math.log2(c / n) for c in cnt.values()) def max_consonant_run(s): vowels set(aeiou) run best 0 for ch in s.lower(): if ch.isalpha() and ch not in vowels: run 1 best max(best, run) else: run 0 return best def domain_features(qname): label qname.split(.)[0] return { length: len(label), entropy: round(shannon_entropy(label), 3), digit_ratio: sum(c.isdigit() for c in label) / max(len(label), 1), consonant_run: max_consonant_run(label), unique_ratio: len(set(label)) / max(len(label), 1), }熵值是最重要的单特征。正常英文单词组成的域名熵值通常在 2.53.5 之间DGA 随机串普遍在 3.8 以上。consonant_run超过 4 的基本可以人工看一眼。digit_ratio对那种纯数字加字母混排的 DGA 特别有效。然后按源 IP 做窗口聚合import pandas as pd df pd.read_json(dns_records.jsonl, linesTrue) df[window] (df[ts] // 300).astype(int) # 5 分钟窗口 df[is_fail] (df[rcode] 3).astype(int) agg df.groupby([src, window]).agg( total(qname, count), fail(is_fail, sum), uniq_domain(qname, nunique), ).reset_index() agg[fail_ratio] agg[fail] / agg[total]fail_ratio超过 0.6 且uniq_domain超过 30 的窗口就是高危窗口。这个阈值不是拍脑袋是我在几个公开数据集上试出来的经验值你可以根据自己环境的基线调整。3.3 规则打分与告警输出把特征和聚合结果合并做一个简单加权打分def risk_score(row): score 0 if row[fail_ratio] 0.6: score 40 if row[uniq_domain] 30: score 30 if row[avg_entropy] 3.8: score 20 if row[max_consonant_run] 5: score 10 return score domain_agg df.groupby(qname).apply( lambda g: pd.Series(domain_features(g.name)) ).reset_index() merged agg.merge( df.groupby([src, window]).apply( lambda g: pd.Series({ avg_entropy: g[qname].apply( lambda x: domain_features(x)[entropy] ).mean(), max_consonant_run: g[qname].apply( lambda x: domain_features(x)[consonant_run] ).max(), }) ).reset_index(), on[src, window] ) merged[score] merged.apply(risk_score, axis1) alerts merged[merged[score] 60] alerts.to_csv(dns_alerts.csv, indexFalse)打分阈值 60 是我一般会用的起点实际部署时建议先跑一周只记录不告警看基线分布再定。输出里保留src、window、score和触发条件方便溯源。4. 避坑与排查DNS 检测里最容易翻车的五件事4.1 把 CDN 域名当成 DGA现象告警里大量出现xxx.cloudfront.net、xxx.akamai.net这类域名熵值高、子域名长。原因CDN 为了负载均衡子域名本身就是随机哈希跟 DGA 特征高度重合。解决维护一个白名单父域名列表匹配到已知 CDN、云厂商、广告域名就直接跳过。白名单不用全覆盖 Top 200 父域名就能消掉大部分误报。4.2 忽略内网 DNS 转发导致的源 IP 失真现象所有告警的源 IP 都是内网 DNS 服务器根本定位不到受控主机。原因内网主机把查询发给本地 DNS本地 DNS 再递归出去你抓的是递归流量。解决要么在本地 DNS 上开查询日志query log要么在交换机做端口镜像抓主机到本地 DNS 的流量。前者更现实BIND 和 dnsmasq 都支持。4.3 时间窗口设得太短或太长现象窗口 1 分钟时告警爆炸窗口 1 小时时什么都抓不到。原因DGA 查询有突发性也有持续性。太短会把正常突发算进去太长会把信号平均掉。解决用双窗口。短窗口 5 分钟抓突发长窗口 1 小时抓持续。两个窗口都超阈值才告警误报率会明显下降。4.4 只看查询不看响应现象检测到大量失败解析但分不清是 DGA 还是用户输错。原因只用了查询侧特征没用响应侧的 TTL、IP 数量、CNAME 链。解决把响应里的ancount、TTL、A 记录数量也纳入特征。DGA 的失败查询通常ancount0而正常错误域名偶尔会有 CNAME 跳转。4.5 忘了 DNS over HTTPS 和 DNS over TLS现象部署后 DNS 日志量骤降以为流量变干净了其实是查询走了加密通道。原因浏览器和部分应用默认开启 DoH/DoT绕过了传统 53 端口。解决在网络层封掉已知 DoH 解析器 IP或者强制内网 DNS 并禁用客户端的 DoH 配置。这一步不做检测覆盖率会打对折。5. 把检测从离线推到在线流式聚合与阈值自学习离线跑通只是第一步。真正要落地得让它在线上持续跑并且阈值能跟着环境变。我一般用river或自己写一个滑动窗口做流式聚合核心是把前面按src window的 groupby 换成增量更新。from collections import defaultdict, deque import time class DNSWindow: def __init__(self, window_sec300): self.window window_sec self.buckets defaultdict(deque) def add(self, src, ts, qname, rcode): key src dq self.buckets[key] dq.append((ts, qname, rcode)) # 淘汰过期记录 cutoff ts - self.window while dq and dq[0][0] cutoff: dq.popleft() def stats(self, src): dq self.buckets[src] if not dq: return None total len(dq) fail sum(1 for _, _, rc in dq if rc 3) uniq len(set(q for _, q, _ in dq)) return { total: total, fail_ratio: fail / total, uniq_domain: uniq, }这个滑动窗口每来一条记录就更新内存占用跟窗口内记录数成正比。线上跑的时候每 10 秒对活跃源 IP 算一次stats超过阈值就推告警。阈值自学习的思路是对每个源 IP 维护过去 7 天的fail_ratio和uniq_domain的均值与标准差用均值 3 * 标准差作为动态阈值。这样新上线的机器不会因为基线不同被误杀老机器突然异常也能被抓住。验证方法上我习惯用两个指标召回率和日均告警量。召回率靠注入已知 DGA 样本测试日均告警量靠线上观察。如果召回率上不去先检查白名单是不是漏了如果告警量太高先调窗口和阈值别急着上模型。我踩过的最大坑是一上来就上 LSTM结果训练数据里正常样本太多模型学了个“全部正常”白折腾两周。后来老老实实做特征工程加规则反而稳定跑了一年多。希望帮到你。本文还有配套的精品资源点击获取
基于DNS流量的僵尸网络检测：从DGA特征到流式告警实战
简介这份资源包面向网络安全初学者与进阶研究者聚焦利用DNS流量分析识别僵尸网络异常行为提供从数据采集、特征提取到模型训练与评估的完整实践方案。包内共27个文件以15个Python脚本为核心覆盖DNS解析、流量抓取、特征工程与检测逻辑另有6个pyc编译文件、2个txt与2个md说明文档、1个readme及1个ipynb笔记本压缩包约3.06MB结构清晰便于按模块查阅。内容涉及查询频率、查询类型、源IP分布、响应延迟等特征并对比支持向量机、随机森林、K-means、Isolation Forest等机器学习方法以及CNN、RNN、Transformer在时序流量上的应用思路。已有313人学习适合希望掌握异常检测流程、复现实验环境并理解DNS安全分析要点的读者参考。1. DNS 流量里的僵尸网络为什么它是最隐蔽的检测入口DNS 是互联网里最不起眼、却最不可能被关掉的协议。一台主机可以没有 HTTP 流量、可以没有 SSH 登录但只要它想跟外界保持联系几乎必然要发 DNS 查询。僵尸网络Botnet的操控者深谙这一点他们把 C2命令与控制指令塞进 DNS 查询名里用 TXT、A、AAAA 记录回传数据甚至用 DNS 隧道把整条 TCP 通道伪装成域名解析。结果就是——防火墙看着 53 端口一切正常流量分析工具却抓不到任何可疑的 HTTP 特征。这个标题要解决的核心问题就是在只有 DNS 流量的前提下把受控主机和恶意域名从海量正常解析里揪出来。它适合两类人一是做企业内网安全的工程师手上有 DNS 日志或镜像流量但不知道怎么下手二是做流量分析 CTF 或安全研究的同学想从 DNS 这个切面建立一套可复现的检测流程。我下面讲的方案不依赖任何商业威胁情报只用公开数据和开源工具从特征提取到规则落地一步步走通。2. 僵尸网络在 DNS 层留下的四类指纹2.1 DGA 域名为什么随机字符串是第一个抓手DGADomain Generation Algorithm是僵尸网络最经典的存活手段。恶意程序内置一个算法每天生成成百上千个候选域名逐个尝试解析其中只有少数几个会被攻击者提前注册并解析到 C2。对检测方来说这带来一个非常稳定的统计特征大量解析失败的随机域名。正常用户也会打错域名但比例极低且错误域名往往接近真实域名编辑距离小。DGA 域名则不同长度集中在 820 字符字符分布接近均匀随机元音辅音比例异常n-gram 得分低。你可以用几个轻量指标先做粗筛指标正常域名典型值DGA 域名典型值域名长度6151025数字占比 10%15%40%最长连续辅音2346唯一字符数 / 长度0.50.70.70.95解析失败率低高这些指标不需要机器学习模型用 Python 十几行就能算。关键是别只看单条查询要看一个时间窗口内的聚合。单次随机域名可能是用户手滑但一台主机在 5 分钟内发起 50 次以上失败解析且域名熵值普遍偏高基本可以判定异常。2.2 DNS 隧道TXT 记录和超长子域名DNS 隧道是把数据编码进域名或 TXT 记录里实现双向通信。检测特征比 DGA 更明确子域名部分异常长单标签超过 50 字符很常见TXT 记录查询占比异常升高同一父域名下出现大量不同子域名查询类型集中为 TXT、NULL、CNAME 等非 A 记录请求/响应包大小分布与正常 DNS 明显不同我一般会先统计每个父域名的子域名基数和平均标签长度。正常 CDN 域名子域名也多但标签长度短且字符集规整隧道域名的子域名往往像 Base64 编码长度整齐、字符集包含大小写字母加数字。2.3 Fast Flux 与域名复用Fast Flux 指同一个域名在短时间内解析到大量不同 IP且 TTL 极短。僵尸网络用这种方式隐藏真实 C2 主机。检测思路是对每个域名统计单位时间内的唯一 A 记录数量和TTL 分布。正常业务域名可能用 CDN但 TTL 不会低到 30 秒以下还频繁换 IP。如果某个域名 10 分钟内解析出 20 个以上不同 IP且 TTL 普遍小于 60 秒就值得拉出来单独看。2.4 心跳与周期性僵尸网络需要定期回连 C2 领取指令这会在 DNS 查询时间序列上留下周期性。正常用户的 DNS 查询跟作息相关白天多、夜间少但不会精确到每 60 秒一次。你可以对每台主机的查询时间戳做差分然后看差分序列的方差。方差极小、间隔稳定的就是心跳特征。3. 用 Python 把 DNS 日志跑成检测流水线3.1 数据准备从 pcap 到结构化记录假设你手上是 pcap 或者已经导出的 DNS 日志。我一般用dnspython加scapy做解析输出成一行一条的 JSON方便后续聚合。先装依赖pip install scapy dnspython pandas numpy解析脚本核心逻辑如下from scapy.all import rdpcap, DNS, DNSQR, DNSRR import json def parse_dns_pcap(pcap_path, out_path): packets rdpcap(pcap_path) records [] for pkt in packets: if pkt.haslayer(DNS): dns pkt[DNS] # 只取查询和响应忽略其他 if dns.qr 0 and dns.qd is not None: qname dns.qd.qname.decode(errorsignore).rstrip(.) qtype dns.qd.qtype records.append({ ts: float(pkt.time), src: pkt[IP].src if pkt.haslayer(IP) else , qname: qname, qtype: qtype, rcode: dns.rcode, ancount: dns.ancount, }) with open(out_path, w) as f: for r in records: f.write(json.dumps(r) \n) parse_dns_pcap(dns.pcap, dns_records.jsonl)这段代码只提取查询报文字段包括时间戳、源 IP、查询名、查询类型、响应码和答案数量。rcode为 3 表示 NXDOMAIN是 DGA 检测的关键字段。ancount为 0 且rcode为 0 的情况也要留意可能是空应答。参数上rdpcap对大文件吃内存超过 500MB 的 pcap 建议先用tcpdump切分tcpdump -r big.pcap -w part_%03d.pcap -C 1003.2 特征计算熵、n-gram 和失败率拿到结构化记录后按「源 IP 时间窗口」聚合。下面这个函数计算单个域名的几个核心特征import math from collections import Counter def shannon_entropy(s): if not s: return 0.0 cnt Counter(s) n len(s) return -sum((c / n) * math.log2(c / n) for c in cnt.values()) def max_consonant_run(s): vowels set(aeiou) run best 0 for ch in s.lower(): if ch.isalpha() and ch not in vowels: run 1 best max(best, run) else: run 0 return best def domain_features(qname): label qname.split(.)[0] return { length: len(label), entropy: round(shannon_entropy(label), 3), digit_ratio: sum(c.isdigit() for c in label) / max(len(label), 1), consonant_run: max_consonant_run(label), unique_ratio: len(set(label)) / max(len(label), 1), }熵值是最重要的单特征。正常英文单词组成的域名熵值通常在 2.53.5 之间DGA 随机串普遍在 3.8 以上。consonant_run超过 4 的基本可以人工看一眼。digit_ratio对那种纯数字加字母混排的 DGA 特别有效。然后按源 IP 做窗口聚合import pandas as pd df pd.read_json(dns_records.jsonl, linesTrue) df[window] (df[ts] // 300).astype(int) # 5 分钟窗口 df[is_fail] (df[rcode] 3).astype(int) agg df.groupby([src, window]).agg( total(qname, count), fail(is_fail, sum), uniq_domain(qname, nunique), ).reset_index() agg[fail_ratio] agg[fail] / agg[total]fail_ratio超过 0.6 且uniq_domain超过 30 的窗口就是高危窗口。这个阈值不是拍脑袋是我在几个公开数据集上试出来的经验值你可以根据自己环境的基线调整。3.3 规则打分与告警输出把特征和聚合结果合并做一个简单加权打分def risk_score(row): score 0 if row[fail_ratio] 0.6: score 40 if row[uniq_domain] 30: score 30 if row[avg_entropy] 3.8: score 20 if row[max_consonant_run] 5: score 10 return score domain_agg df.groupby(qname).apply( lambda g: pd.Series(domain_features(g.name)) ).reset_index() merged agg.merge( df.groupby([src, window]).apply( lambda g: pd.Series({ avg_entropy: g[qname].apply( lambda x: domain_features(x)[entropy] ).mean(), max_consonant_run: g[qname].apply( lambda x: domain_features(x)[consonant_run] ).max(), }) ).reset_index(), on[src, window] ) merged[score] merged.apply(risk_score, axis1) alerts merged[merged[score] 60] alerts.to_csv(dns_alerts.csv, indexFalse)打分阈值 60 是我一般会用的起点实际部署时建议先跑一周只记录不告警看基线分布再定。输出里保留src、window、score和触发条件方便溯源。4. 避坑与排查DNS 检测里最容易翻车的五件事4.1 把 CDN 域名当成 DGA现象告警里大量出现xxx.cloudfront.net、xxx.akamai.net这类域名熵值高、子域名长。原因CDN 为了负载均衡子域名本身就是随机哈希跟 DGA 特征高度重合。解决维护一个白名单父域名列表匹配到已知 CDN、云厂商、广告域名就直接跳过。白名单不用全覆盖 Top 200 父域名就能消掉大部分误报。4.2 忽略内网 DNS 转发导致的源 IP 失真现象所有告警的源 IP 都是内网 DNS 服务器根本定位不到受控主机。原因内网主机把查询发给本地 DNS本地 DNS 再递归出去你抓的是递归流量。解决要么在本地 DNS 上开查询日志query log要么在交换机做端口镜像抓主机到本地 DNS 的流量。前者更现实BIND 和 dnsmasq 都支持。4.3 时间窗口设得太短或太长现象窗口 1 分钟时告警爆炸窗口 1 小时时什么都抓不到。原因DGA 查询有突发性也有持续性。太短会把正常突发算进去太长会把信号平均掉。解决用双窗口。短窗口 5 分钟抓突发长窗口 1 小时抓持续。两个窗口都超阈值才告警误报率会明显下降。4.4 只看查询不看响应现象检测到大量失败解析但分不清是 DGA 还是用户输错。原因只用了查询侧特征没用响应侧的 TTL、IP 数量、CNAME 链。解决把响应里的ancount、TTL、A 记录数量也纳入特征。DGA 的失败查询通常ancount0而正常错误域名偶尔会有 CNAME 跳转。4.5 忘了 DNS over HTTPS 和 DNS over TLS现象部署后 DNS 日志量骤降以为流量变干净了其实是查询走了加密通道。原因浏览器和部分应用默认开启 DoH/DoT绕过了传统 53 端口。解决在网络层封掉已知 DoH 解析器 IP或者强制内网 DNS 并禁用客户端的 DoH 配置。这一步不做检测覆盖率会打对折。5. 把检测从离线推到在线流式聚合与阈值自学习离线跑通只是第一步。真正要落地得让它在线上持续跑并且阈值能跟着环境变。我一般用river或自己写一个滑动窗口做流式聚合核心是把前面按src window的 groupby 换成增量更新。from collections import defaultdict, deque import time class DNSWindow: def __init__(self, window_sec300): self.window window_sec self.buckets defaultdict(deque) def add(self, src, ts, qname, rcode): key src dq self.buckets[key] dq.append((ts, qname, rcode)) # 淘汰过期记录 cutoff ts - self.window while dq and dq[0][0] cutoff: dq.popleft() def stats(self, src): dq self.buckets[src] if not dq: return None total len(dq) fail sum(1 for _, _, rc in dq if rc 3) uniq len(set(q for _, q, _ in dq)) return { total: total, fail_ratio: fail / total, uniq_domain: uniq, }这个滑动窗口每来一条记录就更新内存占用跟窗口内记录数成正比。线上跑的时候每 10 秒对活跃源 IP 算一次stats超过阈值就推告警。阈值自学习的思路是对每个源 IP 维护过去 7 天的fail_ratio和uniq_domain的均值与标准差用均值 3 * 标准差作为动态阈值。这样新上线的机器不会因为基线不同被误杀老机器突然异常也能被抓住。验证方法上我习惯用两个指标召回率和日均告警量。召回率靠注入已知 DGA 样本测试日均告警量靠线上观察。如果召回率上不去先检查白名单是不是漏了如果告警量太高先调窗口和阈值别急着上模型。我踩过的最大坑是一上来就上 LSTM结果训练数据里正常样本太多模型学了个“全部正常”白折腾两周。后来老老实实做特征工程加规则反而稳定跑了一年多。希望帮到你。本文还有配套的精品资源点击获取