简介这份资源面向网络安全初学者与进阶研究者聚焦利用DNS流量分析识别僵尸网络异常行为提供从数据采集、特征提取到模型训练与评估的完整实践方案。压缩包共27个文件约3.06MB以15个Python脚本为核心覆盖DNS解析、PCAP抓取、白名单过滤、阈值判定与流量图谱等模块另含6个pyc编译文件、2个txt与2个md说明文档、1个readme及1个Jupyter Notebook便于直接运行与二次开发。内容围绕查询频率、查询类型、源IP分布、响应延迟等特征展开结合SVM、随机森林、K-means、Isolation Forest等机器学习方法并引入CNN、RNN与Transformer处理DNS时序数据配套环境搭建与可视化流程。目前已有313人学习适合希望掌握异常检测建模、复现实验并理解DNS安全分析思路的读者参考。1. 从一份 DNS 流量日志里揪出僵尸网络这件事到底在做什么手头只有一份 DNS 流量日志怎么判断内网里有没有机器已经被僵尸网络控制这是我在应急响应里被问得最多的问题之一。DNS 是几乎所有恶意软件绕不开的一环——上线要解析 C2 域名DGA 要批量生成随机域名DNS 隧道要往外传数据。相比加密的 HTTP 流量DNS 查询大多以明文形式落在日志里分析成本低、覆盖面广所以基于 DNS 流量分析的僵尸网络检测一直是流量侧检测里性价比很高的一条路。这套方案适合谁安全运营、应急响应、内网威胁狩猎的从业者以及手上有 DNS 日志但不知道怎么下手的运维。它不需要你抓全流量只要拿到递归解析器或 DNS 服务器的查询日志就能跑起来。下面我按「数据从哪来 → 特征怎么提 → 模型怎么判 → 坑在哪」的顺序把这条链路讲透中间给可直接抄的代码和参数。2. DNS 僵尸网络检测的数据底座日志从哪来、字段怎么对齐2.1 三种常见 DNS 日志来源与取舍做检测第一步不是建模是搞清楚你手上有什么数据。DNS 日志的来源直接决定了你能提哪些特征选错了后面全是白干。第一种是递归解析器日志比如 BIND 的 query log、Unbound 的 log-queries、CoreDNS 的 log 插件。这类日志记录的是「谁在什么时候查了什么域名、返回了什么」字段最全是首选。第二种是 DNS 服务器权威日志只能看到外部对你自己域名的查询看不到内网主机查外部域名做僵尸网络检测基本没用。第三种是全流量镜像里解析出来的 DNS 报文用 Zeek、Suricata 或 tcpdump 抓字段最灵活但要自己解析成本最高。我一般优先用递归解析器日志因为它天然覆盖了内网所有主机的出站解析行为。如果你用的是 Ubuntu 22.04 这类系统本地 systemd-resolved 的日志字段偏少建议把内网 DNS 指向自建的 BIND 或 CoreDNS打开 query log 再采集。2.2 把原始日志规整成可分析的表结构原始 query log 是半结构化文本得先规整。以 BIND 的 query log 为例一行长这样# BIND query log 典型格式 26-Mar-2024 10:12:33.123 client 10.0.1.55#54321 (evil-c2.example.com): query: evil-c2.example.com IN A (10.0.0.1)我用一段 Python 把它解析成结构化记录字段包括时间、客户端 IP、查询域名、查询类型、响应码import re import pandas as pd # BIND query log 正则抓时间、客户端、域名、类型、响应标志 PATTERN re.compile( r(?Pts\d{2}-\w{3}-\d{4} \d{2}:\d{2}:\d{2}\.\d{3})\s rclient\s(?Pclient\d\.\d\.\d\.\d)#\d\s r\((?Pqname[^)])\):\squery:\s\S\s(?Pqtype\w)\s r(?Pflags[\-])\s ) def parse_line(line): m PATTERN.search(line) if not m: return None return { ts: m.group(ts), client: m.group(client), qname: m.group(qname).rstrip(.).lower(), qtype: m.group(qtype), flag: m.group(flags), # 表示递归期望- 表示不递归 } records [] with open(query.log, encodingutf-8, errorsignore) as f: for line in f: r parse_line(line) if r: records.append(r) df pd.DataFrame(records) df[ts] pd.to_datetime(df[ts], format%d-%b-%Y %H:%M:%S.%f) df df.sort_values(ts).reset_index(dropTrue) print(df.head())这段代码的逻辑很直白正则把一行日志拆成五个字段qname统一转小写并去掉末尾的点避免Evil.com和evil.com被当成两个域名。flag字段后面有用——正常递归查询是如果内网主机大量发-不递归的查询往往是异常行为。参数上要注意两点一是时间格式%d-%b-%Y依赖日志的英文月份缩写如果你的解析器输出本地化月份得先改 locale 或换正则二是errorsignore是为了防止日志里混入非 UTF-8 字节导致整个文件读挂生产环境建议改成记录坏行而不是静默丢弃。2.3 会话切分按客户端和时间窗聚合单条查询没有意义检测的对象是「一台主机在一段时间内的解析行为」。所以规整完要按客户端 IP 切会话再按时间窗聚合。常见做法是滑动窗口窗口大小 5 到 10 分钟步长 1 分钟。# 按客户端 5 分钟窗口聚合统计查询量、独立域名数等基础指标 df df.set_index(ts) agg ( df.groupby(client) .resample(5min) .agg( q_count(qname, count), # 窗口内查询总数 uniq_domain(qname, nunique), # 独立域名数 nxdomain(flag, lambda s: (s -).sum()), # 近似失败计数 ) .reset_index() ) # 只保留有实际查询的窗口 agg agg[agg[q_count] 0] print(agg.sort_values(q_count, ascendingFalse).head(10))这里resample(5min)是固定窗口工程上更稳的是滑动窗口但固定窗口实现简单、够用。uniq_domain是后面 DGA 检测的核心输入nxdomain这里用 flag 近似真实场景应该解析响应码NXDOMAIN 计数如果你的日志带 rcode 字段就直接用它。提示窗口大小不是拍脑袋定的。太小1 分钟噪声大正常主机的突发解析会被误判太大30 分钟会稀释 DGA 的短时爆发特征。5 到 10 分钟是我在多个内网里验证下来比较平衡的区间。3. 特征工程把 DNS 行为翻译成模型能吃的数字3.1 域名侧特征DGA 域名的三个抓手僵尸网络用 DGA域名生成算法批量造域名这些域名和正常域名在统计上有明显差异。我一般抓三类特征第一类是字符构成。DGA 域名常用随机字母数字组合元音比例偏低、连续辅音偏多、数字占比偏高。第二类是长度和熵。DGA 域名长度分布集中在某个区间比如 12 到 20 字符字符熵明显高于正常域名。第三类是 n-gram 频率正常域名里www、com、常见英文词出现频率高DGA 域名几乎没有。import math from collections import Counter def shannon_entropy(s): # 计算字符串的香农熵DGA 域名熵值通常偏高 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 domain_features(qname): # 取主域名部分去掉 TLD减少 TLD 干扰 parts qname.split(.) sld parts[-2] if len(parts) 2 else qname vowels sum(1 for c in sld if c in aeiou) digits sum(1 for c in sld if c.isdigit()) return { len: len(sld), entropy: round(shannon_entropy(sld), 3), vowel_ratio: round(vowels / max(len(sld), 1), 3), digit_ratio: round(digits / max(len(sld), 1), 3), max_consonant_run: max( (len(run) for run in re.findall(r[^aeiou0-9], sld)), default0 ), } # 对每个窗口内的域名集合求均值/最大值 feat_rows [] for _, row in agg.iterrows(): # 这里简化实际应对窗口内每个域名算特征再聚合 feats domain_features(a8f3k2m9x1q7.com) feats[client] row[client] feats[q_count] row[q_count] feat_rows.append(feats) feat_df pd.DataFrame(feat_rows) print(feat_df.head())shannon_entropy是核心正常域名熵值大多在 2.5 到 3.5 之间DGA 域名经常超过 3.8。max_consonant_run抓的是连续辅音正常英文词很少出现 4 个以上连续辅音DGA 随机串很容易出现。vowel_ratio低于 0.2 也要警惕。参数上sld取倒数第二段是简化处理遇到co.uk这类多级后缀会出错生产环境应该用公共后缀列表Public Suffix List来切。这个坑我在第 5 章会展开。3.2 客户端侧特征查询节奏比内容更能暴露问题光看域名不够僵尸网络主机的查询行为本身就有特征。我重点看四个指标查询频率的突变。正常主机一天查询量相对平稳被感染后上线阶段会出现短时高频查询。独立域名占比。DGA 主机会在短时间内查大量不同域名uniq_domain / q_count比值接近 1。失败率。DGA 生成的域名大部分没注册NXDOMAIN 比例极高正常主机 NXDOMAIN 比例通常在 10% 以下DGA 主机能到 80% 以上。查询类型分布。正常主机以 A、AAAA、HTTPS 为主如果某主机大量发 TXT、NULL、CNAME 查询可能是 DNS 隧道。# 客户端侧特征失败率、域名离散度、查询类型集中度 client_feat agg.groupby(client).agg( total_q(q_count, sum), avg_uniq(uniq_domain, mean), total_nx(nxdomain, sum), ).reset_index() client_feat[nx_ratio] client_feat[total_nx] / client_feat[total_q].clip(lower1) client_feat[domain_dispersion] client_feat[avg_uniq] / client_feat[total_q].clip(lower1) # 按失败率排序快速定位可疑主机 print(client_feat.sort_values(nx_ratio, ascendingFalse).head(10))nx_ratio是最有效的单指标很多场景下光靠它就能筛出八成可疑主机。domain_dispersion接近 1 说明每次查询都是新域名典型的 DGA 行为。这两个指标组合起来比单纯看查询量靠谱得多。3.3 时间侧特征周期性心跳的识别不少僵尸网络有固定心跳比如每 60 秒查一次 C2 域名。这种周期性在时间序列上表现为自相关峰值。做法是把某客户端对某域名的查询时间戳取出来算相邻间隔的方差方差越小越可能是心跳。import numpy as np def heartbeat_score(timestamps): # timestamps: 某客户端对某域名的查询时间列表秒 if len(timestamps) 4: return 0.0 ts np.sort(np.array(timestamps)) intervals np.diff(ts) # 变异系数越小周期性越强 cv intervals.std() / max(intervals.mean(), 1e-6) return round(1 / (1 cv), 3) # 示例对每个 (client, qname) 组合算心跳分 hb ( df.reset_index() .groupby([client, qname])[ts] .apply(lambda s: heartbeat_score(s.astype(int64) // 10**9)) .reset_index(namehb_score) ) print(hb.sort_values(hb_score, ascendingFalse).head(10))heartbeat_score用变异系数取倒数间隔越规律分越高。正常主机的域名查询间隔很随机分数普遍偏低固定心跳的 C2 通信分数能到 0.8 以上。这个特征对识别「低频但规律」的 C2 特别有用因为这类流量查询量不大靠频率阈值抓不到。注意心跳检测对时间精度敏感。如果你的日志时间戳只精确到秒间隔方差会被量化误差放大建议至少毫秒级。BIND 默认毫秒够用。4. 检测模型从规则到无监督怎么选、怎么调4.1 规则基线先跑通再谈模型别一上来就上深度学习。DNS 僵尸网络检测里一套调好的规则能覆盖大部分已知威胁而且可解释、好排查。我一般先建三条基线规则规则一NXDOMAIN 比例超过 60% 且窗口内查询数超过 50。规则二单窗口独立域名数超过 200 且域名平均熵大于 3.5。规则三对同一域名的查询间隔变异系数小于 0.2 且持续超过 10 个周期。# 规则引擎三条基线规则命中即告警 def rule_engine(row): alerts [] if row[nx_ratio] 0.6 and row[total_q] 50: alerts.append(HIGH_NXDOMAIN) if row.get(avg_entropy, 0) 3.5 and row.get(avg_uniq, 0) 200: alerts.append(DGA_LIKE) if row.get(hb_score, 0) 0.8: alerts.append(PERIODIC_C2) return alerts client_feat[alerts] client_feat.apply(rule_engine, axis1) print(client_feat[client_feat[alerts].map(len) 0])规则的好处是每条告警你都能说清楚为什么。HIGH_NXDOMAIN对应 DGA 或域名已失效DGA_LIKE对应随机域名爆发PERIODIC_C2对应心跳通信。阈值不是固定的得按你内网的基线调——先跑一周正常数据看各指标的分布把阈值设在 99 分位附近。4.2 无监督模型Isolation Forest 抓未知变种规则抓已知未知变种得靠无监督。Isolation Forest 在 DNS 异常检测里表现稳定不需要标注数据对高维特征也友好。输入就是前面提的域名侧、客户端侧、时间侧特征拼起来的向量。from sklearn.ensemble import IsolationForest from sklearn.preprocessing import StandardScaler # 组装特征矩阵 feature_cols [nx_ratio, domain_dispersion, avg_uniq, total_q] X client_feat[feature_cols].fillna(0).values scaler StandardScaler() X_scaled scaler.fit_transform(X) # contamination 是预期异常比例按内网规模估 model IsolationForest( n_estimators200, contamination0.05, # 预期 5% 主机异常 random_state42, n_jobs-1, ) client_feat[anomaly] model.fit_predict(X_scaled) # -1 为异常 client_feat[score] model.decision_function(X_scaled) print(client_feat[client_feat[anomaly] -1].sort_values(score))contamination是最关键的参数。设太大误报爆炸设太小漏报。我的经验是先按内网主机数的 3% 到 5% 设跑一段时间后根据告警质量微调。n_estimators200 棵足够再多收益递减。decision_function返回的分数可以排序方便你优先看最异常的。无监督的短板是解释性差所以我的做法是规则和无监督并行规则命中的直接告警无监督命中的进人工复核队列两条链路的结果互相印证。4.3 有监督补充有标注时用 LightGBM如果你手上有历史告警的标注哪些主机确实被感染可以上 LightGBM。它在表格特征上又快又准还能输出特征重要性帮你反推哪些特征真正有用。import lightgbm as lgb from sklearn.model_selection import train_test_split # y 为标注标签1 表示僵尸网络主机 X_train, X_test, y_train, y_test train_test_split( client_feat[feature_cols], client_feat[label], test_size0.3, random_state42 ) clf lgb.LGBMClassifier( n_estimators300, learning_rate0.05, num_leaves31, scale_pos_weight10, # 正负样本不平衡时调高 random_state42, ) clf.fit(X_train, y_train) # 特征重要性指导后续特征裁剪 for name, imp in sorted(zip(feature_cols, clf.feature_importances_), keylambda x: -x[1]): print(f{name}: {imp})scale_pos_weight是处理样本不平衡的关键僵尸网络主机在整体里是极少数不调这个参数模型会偏向多数类。num_leaves31 是默认值特征维度不高时够用。特征重要性输出后如果某个特征贡献极低可以考虑裁掉减少计算开销。5. 避坑与排查DNS 检测里最容易翻车的五件事5.1 现象误报集中在办公网段每天几百条原因办公网主机装了各种软件后台自动更新、CDN 探测、遥测上报都会产生大量域名查询NXDOMAIN 比例和域名离散度天然偏高直接套阈值必然误报。解决按网段分组建基线办公网、服务器网段、IoT 网段分开设阈值。办公网的 NXDOMAIN 阈值可以放宽到 70%服务器网段收紧到 40%。另外把已知的软件更新域名、CDN 域名加白名单白名单要定期更新别一劳永逸。5.2 现象DGA 检测对某些家族完全失效原因不是所有 DGA 都靠随机字符。有些家族用字典拼接比如apple-support-login.com字符熵和正常域名没区别靠熵值抓不到。解决对字典型 DGA得换特征——看域名和已知品牌词的编辑距离、看子域名层级深度、看注册时间如果日志带 whois 信息。单靠字符统计覆盖不了所有家族这也是为什么规则和无监督要并行。5.3 现象公共后缀切分错误导致特征失真原因sld parts[-2]这种简化切分遇到example.co.uk会把co当成主域名长度、熵全算错特征直接废掉。解决用公共后缀列表Public Suffix List来切。Python 里可以用tldextract库它会正确处理多级后缀import tldextract def get_sld(qname): # tldextract 正确处理 co.uk、com.cn 等多级后缀 ext tldextract.extract(qname) return ext.domain # 返回真正的主域名部分 print(get_sld(login.example.co.uk)) # 输出 exampletldextract首次运行会下载后缀列表离线环境要提前把列表文件打包进去否则会卡住。5.4 现象DNS 隧道检测把正常 TXT 查询也告警了原因SPF、DKIM、DMARC 校验都会发 TXT 查询邮件量大的内网 TXT 查询本来就多光看查询类型会误判。解决DNS 隧道的特征是「TXT 查询的响应数据量大且编码异常」不是查询类型本身。要结合响应长度和内容熵来判断光看 qtype 不够。如果你的日志没有响应长度字段就得从全流量里补或者退而求其次看 TXT 查询的目标域名是否集中且陌生。5.5 现象模型上线后效果衰减几个月后告警质量下降原因僵尸网络在进化DGA 算法在变C2 基础设施在换静态模型必然衰减。这不是 bug是这类检测的固有特性。解决建立定期重训机制至少每月用新数据重跑一次无监督模型每季度复核一次规则阈值。同时保留人工复核的反馈闭环把确认的误报和漏报回灌到训练集。没有反馈闭环的检测系统半年后基本就废了。6. 把检测跑成常态验证方法与一个我常用的技巧检测系统建好只是开始怎么验证它真的有用比建模本身更重要。我常用的验证方法是「注入测试」在隔离环境里用一台测试机模拟 DGA 查询和心跳通信看检测链路能不能在预期时间内告警。具体做法是用脚本按固定间隔查询一批随机生成的域名然后检查规则和无监督模型是否命中。import random import string import time import socket def gen_dga_domain(): # 模拟 DGA随机 12 位字母数字 固定 TLD s .join(random.choices(string.ascii_lowercase string.digits, k12)) return f{s}.test-dga.invalid # 在隔离测试机上运行模拟 DGA 爆发 for _ in range(300): domain gen_dga_domain() try: socket.gethostbyname(domain) # 触发解析产生日志 except socket.gaierror: pass # 域名不存在正常 time.sleep(0.05)这段脚本会在短时间内产生 300 条 NXDOMAIN 查询如果检测链路正常对应测试机的nx_ratio会飙升规则应该命中HIGH_NXDOMAIN。用.invalid顶级域是为了保证域名一定解析失败同时不污染真实 DNS。注入测试要定期做尤其是模型重训或规则调整之后确认链路没断。另一个我踩过坑才养成的习惯所有告警必须带上下文快照。光告警「某主机 NXDOMAIN 比例高」没用得同时存下触发告警的那个时间窗内的原始查询样本至少 50 条。这样人工复核时不用回头翻日志直接看快照就能判断是真感染还是误报。这个习惯帮我省了大量排查时间也让误报分析变得可追溯。最后说个心态上的教训。我早期做 DNS 检测时总想一步到位追求低误报低漏报结果调参调到怀疑人生。后来想通了DNS 检测的价值不在于单点精准而在于它是一个低成本、广覆盖的筛子把可疑主机从几千台里筛到几十台剩下的交给人工和主机侧检测。接受一定误报把精力放在告警排序和复核效率上整条链路才跑得起来。希望帮到你。本文还有配套的精品资源点击获取