| name | detecting-dns-exfiltration-with-dns-query-analysis |
| description | 通过分析查询熵值、子域名长度、查询量、TXT 记录滥用及响应载荷大小，使用被动 DNS 监控检测 DNS 隧道数据渗出。 |
| domain | cybersecurity |
| subdomain | network-security |
| tags | ["dns-exfiltration","dns-tunneling","data-exfiltration","threat-detection","entropy-analysis","passive-dns","network-monitoring","iodine","dnscat2"] |
| version | 1.0 |
| author | mahipal |
| license | Apache-2.0 |
通过 DNS 查询分析检测 DNS 渗出
概述
DNS 渗出（DNS exfiltration）将域名系统（Domain Name System）作为隐蔽通道，从被入侵的网络中提取数据。攻击者将窃取的数据编码到 DNS 查询名称（子域名）或 DNS 响应记录（TXT、CNAME、NULL）中，绕过通常对 DNS 流量放行的传统安全控制。iodine、dnscat2 和 dns2tcp 等工具可实现通过 DNS 的完整 TCP 隧道。检测需要分析 DNS 查询模式的异常，包括：过长的查询长度、高熵值子域名字符串、对单一域名的异常查询量以及超大 TXT 记录响应。本技能介绍如何使用被动 DNS（passive DNS）分析、统计方法和机器学习方法构建全面的 DNS 渗出检测能力。
前置条件
- 可访问 DNS 查询日志（被动 DNS 抓包、DNS 服务器日志或 PCAP）
- Zeek、Suricata 或 tcpdump 用于 DNS 流量捕获
- Python 3.8+，需安装 scipy、numpy、pandas 和 scikit-learn
- SIEM 平台用于告警关联
- 已建立环境正常 DNS 流量模式的基线
核心概念
DNS 隧道原理
DNS 渗出在 DNS 消息的不同部分对数据进行编码：
出站（基于查询的渗出）：
将编码数据作为子域名标签：
dGhlIHNlY3JldCBkYXRh.exfil.attacker.com
[base64 编码数据].[隧道域名]
使用的查询类型：A、AAAA、CNAME、MX、TXT、NULL
入站（基于响应的命令通道）：
TXT 记录在响应中携带编码的命令/数据
CNAME 记录通过多个标签链式传递编码数据
NULL 记录携带任意二进制数据
检测指标
| 指标 | 正常 DNS | DNS 隧道 |
|---|
| 子域名长度 | 5-20 字符 | 40-253 字符 |
| 标签数量 | 2-4 个 | 5-10+ 个 |
| 香农熵（Shannon entropy） | 2.5-3.5 bits | 4.0-5.5 bits |
| 查询量（每域名） | 可变 | 每分钟数百至数千 |
| TXT 响应大小 | < 100 字节 | 200-4000+ 字节 |
| 唯一子域名 | 少 | 极多 |
| 查询类型分布 | 以 A/AAAA 为主 | 大量 TXT、NULL、CNAME |
常见隧道工具
| 工具 | 协议 | 编码方式 | 检测难度 |
|---|
| iodine | IP-over-DNS | Base32/Base64/Raw | 中 |
| dnscat2 | TCP-over-DNS | 十六进制编码 | 中 |
| dns2tcp | TCP-over-DNS | Base64 | 中 |
| DNSExfiltrator | 自定义 | Base64 | 低 |
| Cobalt Strike DNS | C2 over DNS | 自定义编码 | 高 |
实施步骤
步骤 1：捕获 DNS 流量
使用 Zeek：
zeek -i eth0 -C base/protocols/dns
zeek -r traffic.pcap base/protocols/dns
使用 tcpdump：
tcpdump -i eth0 -w dns_capture.pcap port 53
tcpdump -i eth0 -w large_dns.pcap 'port 53 and greater 512'
使用 Suricata：
outputs:
- eve-log:
types:
- dns:
query: yes
answer: yes
formats: [detailed]
步骤 2：分析查询特征
用于 DNS 渗出检测的 Python 脚本：
"""DNS 渗出检测器——分析 DNS 日志中的隧道指标。"""
import json
import math
import re
import sys
from collections import defaultdict
from datetime import datetime, timedelta
import pandas as pd
def calculate_entropy(domain: str) -> float:
"""计算字符串的香农熵。"""
if not domain:
return 0.0
freq = defaultdict(int)
for char in domain:
freq[char] += 1
length = len(domain)
entropy = -sum(
(count / length) * math.log2(count / length)
for count in freq.values()
)
return entropy
def extract_subdomain(query: str) -> str:
"""从 FQDN 中提取子域名部分。"""
parts = query.rstrip('.').split('.')
if len(parts) > 2:
return '.'.join(parts[:-2])
return ''
def get_base_domain(query: ) -> :
parts = query.rstrip().split()
(parts) >= :
.join(parts[-:])
query
() -> :
b64_chars = ()
(s) < :
char_ratio = ( c s c b64_chars) / (s)
char_ratio > calculate_entropy(s) >
() -> :
hex_chars = ()
(s) < :
clean = s.replace(, ).replace(, )
(c hex_chars c clean) (clean) % ==
:
():
.domain_stats = defaultdict(: {
: ,
: (),
: ,
: ,
: defaultdict(),
: (),
: ,
: ,
: [],
})
.thresholds = {
: ,
: ,
: ,
: ,
: ,
: ,
: ,
: ,
}
():
base_domain = get_base_domain(query)
subdomain = extract_subdomain(query)
stats = .domain_stats[base_domain]
stats[] +=
stats[].add(subdomain)
stats[] += (subdomain)
stats[] += calculate_entropy(subdomain)
stats[][qtype] +=
stats[].add(src_ip)
stats[] :
stats[] = timestamp
stats[] = timestamp
qtype (, ) response_size > :
stats[].append(response_size)
():
alerts = []
domain, stats .domain_stats.items():
stats[] < .thresholds[]:
unique_count = (stats[])
avg_length = stats[] / stats[]
avg_entropy = stats[] / stats[]
unique_ratio = unique_count / stats[]
txt_queries = stats[].get(, ) + stats[].get(, )
txt_ratio = txt_queries / stats[]
score =
indicators = []
avg_length > .thresholds[]:
score +=
indicators.append()
avg_entropy > .thresholds[]:
score +=
indicators.append()
unique_ratio > .thresholds[]:
score +=
indicators.append()
txt_ratio > .thresholds[]:
score +=
indicators.append()
unique_count > .thresholds[]:
score +=
indicators.append()
encoded_count = (
sd (stats[])[:]
is_base64_like(sd) is_hex_encoded(sd)
)
encoded_count > :
score +=
indicators.append()
score >= :
duration = (stats[] - stats[]).total_seconds() stats[] stats[]
alerts.append({
: domain,
: (score, ),
: stats[],
: unique_count,
: (avg_length, ),
: (avg_entropy, ),
: (unique_ratio, ),
: (txt_ratio, ),
: (stats[]),
: duration,
: indicators,
})
(alerts, key= x: x[], reverse=)
():
(log_path, ) f:
line f:
line.startswith():
fields = line.strip().split()
(fields) < :
:
ts = datetime.fromtimestamp((fields[]))
src_ip = fields[]
query = fields[]
qtype = fields[]
.process_query(ts, src_ip, query, qtype)
(ValueError, IndexError):
():
(log_path, ) f:
line f:
:
event = json.loads(line)
event.get() != :
dns = event.get(, {})
ts = datetime.fromisoformat(event[].replace(, ))
src_ip = event.get(, )
query = dns.get(, )
qtype = dns.get(, )
.process_query(ts, src_ip, query, qtype)
(json.JSONDecodeError, KeyError, ValueError):
():
detector = DNSExfiltrationDetector()
log_file = sys.argv[] (sys.argv) >
log_file.endswith():
detector.process_eve_json(log_file)
:
detector.process_zeek_dns_log(log_file)
alerts = detector.analyze()
alerts:
()
()
()
alert alerts:
severity = alert[] >= alert[] >=
()
()
()
()
()
()
()
:
()
__name__ == :
main()
步骤 3：部署 Suricata 规则检测 DNS 渗出
# 检测过长 DNS 查询（潜在隧道）
alert dns $HOME_NET any -> any 53 (msg:"DNS Exfiltration - Excessive query length"; dns.query; content:"."; pcre:"/^.{60,}/"; threshold:type both,track by_src,count 20,seconds 60; classtype:bad-unknown; sid:3000001; rev:1;)
# 检测高熵值 DNS 子域名
alert dns $HOME_NET any -> any 53 (msg:"DNS Exfiltration - High entropy subdomain"; dns.query; pcre:"/^[a-zA-Z0-9+\/=]{30,}\./"; threshold:type both,track by_src,count 10,seconds 60; classtype:bad-unknown; sid:3000002; rev:1;)
# 检测大型 TXT 记录响应
alert dns any 53 -> $HOME_NET any (msg:"DNS Exfiltration - Large TXT response"; content:"|00 10|"; byte_test:2,>,400,0,relative; classtype:bad-unknown; sid:3000003; rev:1;)
# 检测 NULL 记录查询（iodine 指标）
alert dns $HOME_NET any -> any 53 (msg:"DNS Exfiltration - NULL record query (iodine indicator)"; content:"|00 0a|"; classtype:bad-unknown; sid:3000004; rev:1;)
# 检测 dnscat2 流量模式
alert dns $HOME_NET any -> any 53 (msg:"DNS Exfiltration - dnscat2 indicator"; dns.query; content:"dnscat"; nocase; classtype:trojan-activity; sid:3000005; rev:1;)
步骤 4：SIEM 检测规则
Splunk SPL 查询用于 DNS 渗出检测：
index=dns sourcetype=zeek:dns
| eval subdomain=mvindex(split(query,"."),0)
| eval subdomain_len=len(subdomain)
| eval label_count=mvcount(split(query,"."))
| stats count as query_count,
dc(subdomain) as unique_subdomains,
avg(subdomain_len) as avg_sub_len,
values(src_ip) as source_ips
by query_domain
| where query_count > 100 AND avg_sub_len > 30 AND unique_subdomains > 50
| eval risk_score = case(
avg_sub_len > 50 AND unique_subdomains > 200, "Critical",
avg_sub_len > 40 AND unique_subdomains > 100, "High",
avg_sub_len > 30 AND unique_subdomains > 50, "Medium",
true(), "Low")
| sort -query_count
| table query_domain risk_score query_count unique_subdomains avg_sub_len source_ips
响应操作
- 封锁隧道域名——在 DNS 解析器和防火墙层面封锁
- 隔离源主机——从网络中隔离以进行取证调查
- 捕获完整 PCAP——保留 DNS 流量作为证据
- 识别渗出数据——解码捕获的 DNS 查询以还原数据
- 检查持久化机制——排查被入侵主机上的持久化手段
- 更新阻断名单——将识别出的 C2 域名和基础设施加入封锁列表
最佳实践
- DNS 日志记录——在解析器和网络层面启用完整的 DNS 查询和响应日志
- 仅使用内部 DNS——强制所有 DNS 通过内部解析器；封锁直接外部 DNS（端口 53）
- 响应策略区域（RPZ）——部署 RPZ feeds 封锁已知隧道域名
- 先建立基线——在设置检测阈值前，先建立正常 DNS 查询模式的基线
- TXT 记录监控——重点关注 TXT 和 NULL 记录查询
- 加密 DNS 感知——监控可能绕过 DNS 检测的 DoH/DoT 使用情况
参考资料
- Splunk DNS 渗出深度学习检测
- Akamai DNS 数据渗出
- SANS 检测 DNS 隧道
- Fidelis DNS 隧道检测
