# Windows - 检查hosts文件内容
type C:\Windows\System32\drivers\etc\hosts

# Linux/macOS - 检查hosts文件修改时间
stat /etc/hosts

# 对比hosts文件哈希与基线值
certutil -hashfile C:\Windows\System32\drivers\etc\hosts SHA256

# 使用diff对比基线版本（假设有备份）
diff /etc/hosts /var/backup/hosts.baseline

# Linux/macOS - 检查当前系统DNS配置
scutil --dns | grep 'nameserver'

# Windows - 检查网络接口DNS配置
ipconfig /all | findstr "DNS"

# 检查路由器默认DNS是否被篡改（通过SNMP或API）
dig @8.8.8.8 google.com +short

# 对比dig结果与本地DNS解析结果
dig google.com +short
nslookup google.com 2>&1 | grep Address

# 检查DNS缓存记录
# Windows
ipconfig /displaydns

# Linux（使用systemd-resolved）
resolvectl statistics
resolvectl query example.com

# Linux（使用dnsmasq）
kill -USR1 $(pidof dnsmasq)  # 向dnsmasq发送信号获取统计

# 检查DNS响应中的异常TTL值
dig +nocmd +nocomment +noquestion example.com any | grep -E "^[a-z]"

# 使用dnstop监控实时DNS流量
dnstop -s eth0

# 检查BGP路由变动历史
bgpdump /var/log/bgp/updates.log | grep -E "2026-07-.*103.102.100"

# 通过公开BGP监控API查询AS路径
curl -s "https://stat.ripe.net/data/bgplay/data.json?resource=103.102.100.0/24"

# 对比多地区的DNS解析结果
# 使用Python脚本跨区域查询
python3 -c "
import dns.resolver
import json
providers = {
    'google': '8.8.8.8',
    'cloudflare': '1.1.1.1',
    'opendns': '208.67.222.222',
    'quad9': '9.9.9.9'
}
results = {}
target = 'paypal.com'
for name, server in providers.items():
    resolver = dns.resolver.Resolver()
    resolver.nameservers = [server]
    answers = resolver.resolve(target, 'A')
    results[name] = [str(r) for r in answers]
print(json.dumps(results, indent=2))
"

import dpkt
import socket
from collections import defaultdict, Counter
import math
import statistics

def calculate_entropy(data):
    if not data:
        return 0
    entropy = 0
    for x in range(256):
        p_x = data.count(x) / len(data)
        if p_x > 0:
            entropy += -p_x * math.log2(p_x)
    return entropy

def extract_dns_queries(pcap_file):
    queries = []
    with open(pcap_file, 'rb') as f:
        pcap = dpkt.pcap.Reader(f)
        for ts, buf in pcap:
            try:
                eth = dpkt.ethernet.Ethernet(buf)
                ip = eth.data
                if isinstance(ip, dpkt.ip.IP) and ip.p == dpkt.ip.IP_PROTO_UDP:
                    udp = ip.data
                    if udp.dport == 53 or udp.sport == 53:
                        dns = dpkt.dns.DNS(udp.data)
                        if len(dns.qd) > 0:
                            q = dns.qd[0]
                            qname = q.name.decode('utf-8', errors='ignore')
                            qtype = q.type
                            src_ip = socket.inet_ntoa(ip.src)
                            dst_ip = socket.inet_ntoa(ip.dst)
                            queries.append({
                                'timestamp': ts,
                                'src_ip': src_ip,
                                'dst_ip': dst_ip,
                                'qname': qname,
                                'qtype': qtype,
                                'length': len(buf)
                            })
            except Exception:
                continue
    return queries

def detect_dns_tunnel(queries, threshold_entropy=4.5, threshold_freq=100):
    alerts = []
    qname_entropy = {}
    domain_count = Counter()
    qtype_dist = Counter()
    length_dist = []

    for q in queries:
        labels = q['qname'].rstrip('.').split('.')
        if len(labels) >= 3:
            subdomain = '.'.join(labels[:-2])
            entropy = calculate_entropy(subdomain.encode())
            qname_entropy[q['qname']] = entropy
        domain_count[q['src_ip']] += 1
        qtype_dist[(q['src_ip'], q['qtype'])] += 1
        length_dist.append(q['length'])

    for src_ip, count in domain_count.most_common():
        if count > threshold_freq:
            high_entropy = []
            for q in queries:
                if q['src_ip'] == src_ip:
                    labels = q['qname'].rstrip('.').split('.')
                    if len(labels) >= 3:
                        subdomain = '.'.join(labels[:-2])
                        ent = calculate_entropy(subdomain.encode())
                        if ent > threshold_entropy:
                            high_entropy.append(q['qname'])
            if len(high_entropy) > count * 0.3:
                alerts.append({
                    'src_ip': src_ip,
                    'query_count': count,
                    'high_entropy_ratio': len(high_entropy) / count,
                    'sample_queries': high_entropy[:5],
                    'avg_length': statistics.mean(length_dist) if length_dist else 0
                })

    return alerts

if __name__ == '__main__':
    import sys
    pcap_file = sys.argv[1] if len(sys.argv) > 1 else 'capture.pcap'
    queries = extract_dns_queries(pcap_file)
    alerts = detect_dns_tunnel(queries)
    print(f'Total queries: {len(queries)}')
    print(f'DNS tunnel alerts: {len(alerts)}')
    for alert in alerts:
        print(f'  Source: {alert["src_ip"]}')
        print(f'  Queries: {alert["query_count"]}')
        print(f'  High entropy ratio: {alert["high_entropy_ratio"]:.2%}')
        print(f'  Sample: {alert["sample_queries"][:2]}')

# 检查子域名长度异常
tcpdump -r traffic.pcap -nn port 53 | awk '{print $NF}' | \
  awk -F. '{print $1}' | awk '{print length}' | sort -n | tail -10

# 统计TXT查询占比
tcpdump -r traffic.pcap -nn port 53 | grep "TXT" | wc -l

# 查看特定域名下的TXT查询
tshark -r traffic.pcap -Y "dns.qry.type == 16" -T fields -e dns.qry.name | \
  sort | uniq -c | sort -rn | head -20

import datetime
import hashlib
import argparse
import requests

def dga_conficker_b(date_str=None):
    if date_str is None:
        date = datetime.datetime.now()
    else:
        date = datetime.datetime.strptime(date_str, '%Y-%m-%d')
    year = date.year
    month = date.month
    day = date.day
    tlds = ['com', 'net', 'org', 'info', 'biz', 'ws', 'us', 'cc']
    domains = []
    for tld in tlds:
        seed = f'{year}{month:02d}{day:02d}{tld}'
        for i in range(250):
            data = f'{seed}{i}'
            md5 = hashlib.md5(data.encode()).hexdigest()
            length = (int(md5[:2], 16) % 6) + 4
            domain_chars = []
            for j in range(length):
                char_val = int(md5[2+j*2:4+j*2], 16) % 36
                if char_val < 26:
                    domain_chars.append(chr(ord('a') + char_val))
                else:
                    domain_chars.append(chr(ord('0') + char_val - 26))
            domain = ''.join(domain_chars)
            domains.append(f'{domain}.{tld}')
    return domains

def dga_trickbot(date_str=None):
    if date_str is None:
        date = datetime.datetime.now()
    else:
        date = datetime.datetime.strptime(date_str, '%Y-%m-%d')
    year = date.year
    month = date.month
    day = date.day
    domains = []
    tlds = ['com', 'net', 'org', 'info']
    charset = 'abcdefghijklmnopqrstuvwxyz0123456789'
    for tld in tlds:
        seed = f'{year}{month:02d}{day:02d}{tld}'
        for i in range(100):
            data = seed + str(i) * 3
            md5 = hashlib.md5(data.encode()).hexdigest()
            length = (int(md5[-4:], 16) % 16) + 12
            domain = ''
            for j in range(length):
                idx = int(md5[j*2:(j+1)*2], 16) % len(charset)
                domain += charset[idx]
            domains.append(f'{domain}.{tld}')
    return domains

def check_domain_availability(domains, sample_size=100):
    import socket
    import concurrent.futures
    results = {'resolved': [], 'nxdomain': [], 'error': []}
    test_domains = domains[:sample_size]

    def check(domain):
        try:
            ip = socket.gethostbyname(domain)
            return (domain, 'resolved', ip)
        except socket.gaierror:
            return (domain, 'nxdomain', None)
        except Exception as e:
            return (domain, 'error', str(e))

    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        for result in executor.map(check, test_domains):
            domain, status, extra = result
            results[status].append((domain, extra))
    return results

def compute_ngram_features(domains):
    from collections import Counter
    bigrams = Counter()
    trigrams = Counter()
    lengths = []
    for domain in domains:
        name = domain.split('.')[0]
        lengths.append(len(name))
        for i in range(len(name) - 1):
            bigrams[name[i:i+2]] += 1
        for i in range(len(name) - 2):
            trigrams[name[i:i+3]] += 1
    total_bigrams = sum(bigrams.values())
    total_trigrams = sum(trigrams.values())
    return {
        'avg_length': sum(lengths) / len(lengths),
        'unique_bigram_ratio': len(bigrams) / total_bigrams if total_bigrams > 0 else 1,
        'unique_trigram_ratio': len(trigrams) / total_trigrams if total_trigrams > 0 else 1,
        'top_bigrams': bigrams.most_common(5)
    }

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='DGA Domain Generator & Analyzer')
    parser.add_argument('--family', choices=['conficker', 'trickbot'], default='conficker')
    parser.add_argument('--date', help='Date in YYYY-MM-DD format')
    parser.add_argument('--check', action='store_true', help='Check domain resolution')
    args = parser.parse_args()

    if args.family == 'conficker':
        domains = dga_conficker_b(args.date)
    else:
        domains = dga_trickbot(args.date)

    print(f'Generated {len(domains)} domains for {args.family}')
    features = compute_ngram_features(domains)
    print(f'Average length: {features["avg_length"]:.1f}')
    print(f'Bigram uniqueness: {features["unique_bigram_ratio"]:.3f}')
    print(f'Top bigrams: {features["top_bigrams"]}')

    if args.check:
        results = check_domain_availability(domains)
        print(f'Resolved: {len(results["resolved"])}')
        print(f'NXDOMAIN: {len(results["nxdomain"])}')
        for domain, ip in results['resolved'][:10]:
            print(f'  {domain} -> {ip}')

title: DGA Domain NXDOMAIN Flood Detection
id: bf8c3a21-e25d-4a5c-9f17-8b4d5e6f7a8b
status: experimental
description: Detects high volume of NXDOMAIN responses from a single client, indicating potential DGA domain generation and connection attempts
author: BlueTeam Analyst
date: 2026-07-08
tags:
  - attack.t1568.002
  - attack.ta0011
logsource:
  category: dns
  product: windows
  service: dns-server
detection:
  selection:
    EventID: 264
    RCODE: 3
  timeframe: 5m
  condition:
    selection | count() > 100 by SourceIp
falsepositives:
  - Misconfigured applications
  - Vulnerability scanners
  - DNS resolver testing
level: medium

# Windows - 通过组策略禁用DoH
# 设置: Computer Configuration > Administrative Templates > Network > DNS Client
# 策略: "Turn off DoH" -> Enabled

# macOS - 通过配置文件禁用DoH
# /etc/dnssec/dns.conf 配置

# Linux - 本地DNS代理强制重定向
iptables -t nat -A OUTPUT -p udp --dport 853 -j REDIRECT --to-port 5353
iptables -t nat -A OUTPUT -p tcp --dport 853 -j REDIRECT --to-port 5353

# 使用dnsmasq作为本地DNS代理
cat > /etc/dnsmasq.d/block-doh.conf << 'DNSMASQ_CONF'
# 拦截已知DoH服务器域名
server=/cloudflare-dns.com/0.0.0.0
server=/mozilla.cloudflare-dns.com/0.0.0.0
server=/dns.google/0.0.0.0
server=/dns.quad9.net/0.0.0.0
server=/dns.adguard.com/0.0.0.0
DNSMASQ_CONF

# 检测网络中DoH流量（SNI分析）
tcpdump -ni eth0 -A 'tcp port 443' | grep -E "cloudflare-dns|dns.google|dns.quad9"

# 使用tshark提取TLS Client Hello指纹
tshark -r capture.pcap -Y "tls.handshake.type == 1" -T fields \
  -e tls.handshake.ciphersuite \
  -e tls.handshake.extensions_server_name \
  -e tls.handshake.extensions_supported_group

# JA3指纹匹配（通过Python）
python3 -c "
import json
doh_fingerprints = {
    '785f5b62c12521a11a10a31a8d5d6c89': 'Firefox DoH',
    '51c2f0e5e7e2cc7a8e7c9f3b1a2d4e6f': 'Chrome DoH',
    'a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6': 'curl DoH'
}
target = 'your_ja3_here'
if target in doh_fingerprints:
    print(f'Detected: {doh_fingerprints[target]}')
else:
    print('Unknown fingerprint, may need analysis')
"

import json
import requests
import datetime
from collections import defaultdict, Counter

class PassiveDNSAnalyzer:
    def __init__(self, virustotal_api_key=None):
        self.api_key = virustotal_api_key
        self.dns_history = defaultdict(list)

    def query_pdns(self, domain):
        query_url = f'https://otx.alienvault.com/api/v1/indicators/domain/{domain}/passive_dns'
        try:
            resp = requests.get(query_url)
            if resp.status_code == 200:
                records = resp.json().get('passive_dns', [])
                for record in records:
                    self.dns_history[domain].append({
                        'hostname': record.get('hostname'),
                        'address': record.get('address'),
                        'first_seen': record.get('first_seen'),
                        'last_seen': record.get('last_seen'),
                        'record_type': record.get('record_type')
                    })
                return True
        except Exception as e:
            print(f'Query failed for {domain}: {e}')
        return False

    def analyze_dns_stability(self):
        analysis = {}
        for domain, records in self.dns_history.items():
            ip_changes = []
            unique_ips = set()
            for r in records:
                if r['record_type'] in ['A', 'AAAA']:
                    if r['address']:
                        unique_ips.add(r['address'])
                        ip_changes.append({
                            'ip': r['address'],
                            'time': r.get('first_seen', r.get('last_seen'))
                        })
            ip_count = len(unique_ips)
            if ip_count > 0:
                ip_changes_sorted = sorted(ip_changes, key=lambda x: x['time'] or '')
                period_days = 0
                if len(ip_changes_sorted) > 1:
                    try:
                        t1 = datetime.datetime.fromisoformat(ip_changes_sorted[0]['time'])
                        t2 = datetime.datetime.fromisoformat(ip_changes_sorted[-1]['time'])
                        period_days = (t2 - t1).days
                    except Exception:
                        pass
                change_rate = ip_count / max(period_days, 1)
                analysis[domain] = {
                    'unique_ips': ip_count,
                    'ip_change_rate': f'{change_rate:.4f}/day',
                    'period_days': period_days,
                    'is_stable': ip_count <= 3
                }
        return analysis

    def find_shared_infrastructure(self, domains):
        ip_domain_map = defaultdict(list)
        for domain in domains:
            for record in self.dns_history.get(domain, []):
                if record['record_type'] in ['A', 'AAAA'] and record.get('address'):
                    ip_domain_map[record['address']].append(domain)
        shared = {}
        for ip, doms in ip_domain_map.items():
            if len(doms) > 1:
                shared[ip] = list(set(doms))
        return shared

    def cluster_by_ns_or_mx(self, domains):
        ns_cluster = defaultdict(list)
        mx_cluster = defaultdict(list)
        for domain in domains:
            for record in self.dns_history.get(domain, []):
                if record['record_type'] == 'NS':
                    ns_cluster[record['address']].append(domain)
                elif record['record_type'] == 'MX':
                    mx_cluster[record['address']].append(domain)
        return {
            'ns_clusters': {k: list(set(v)) for k, v in ns_cluster.items() if len(set(v)) > 1},
            'mx_clusters': {k: list(set(v)) for k, v in mx_cluster.items() if len(set(v)) > 1}
        }

if __name__ == '__main__':
    analyst = PassiveDNSAnalyzer()
    test_domains = ['malware1.com', 'malware2.net', 'malware3.org']
    for domain in test_domains:
        analyst.query_pdns(domain)
    stability = analyst.analyze_dns_stability()
    shared = analyst.find_shared_infrastructure(test_domains)
    clusters = analyst.cluster_by_ns_or_mx(test_domains)
    print('DNS Stability Analysis:')
    for domain, info in stability.items():
        status = 'STABLE' if info['is_stable'] else 'UNSTABLE'
        print(f'  {domain}: {info["unique_ips"]} IPs, {status}')
    print(f'Shared infrastructure: {len(shared)} IPs')

# 提取DNS日志中的NXDOMAIN统计
cat dns.log | awk -F'\t' '$10 == 3 {print $7}' | \
  sort | uniq -c | sort -rn | head -20

# DNS日志与NetFlow关联（查询特定时间窗口的DNS和IP连接记录）
awk -F'\t' '$1 ~ "2026-07-08 0[0-5]:" {print $7, $9}' dns.log > dns_queries.txt
awk '{print $1}' dns_queries.txt | sort -u > resolved_ips.txt
grep -Ff resolved_ips.txt netflow.log | awk '{print $1, $3, $5}' | \
  sort | uniq -c | sort -rn | head -20

# 检测速变域名（Fast-Flux）
awk -F'\t' '$7 ~ /^[a-z0-9]{20,}\./ {print $1, $7, $9}' dns.log | \
  sort -k2 | uniq -c | sort -rn | head -30

# 检查TLS SNI与HTTP Host不一致性
tshark -r capture.pcap -Y "tls.handshake.type == 1" -T fields \
  -e tls.handshake.extensions_server_name | sort -u > sni_hosts.txt

tshark -r capture.pcap -Y "http.request" -T fields \
  -e http.host | sort -u > http_hosts.txt

# 比对两者差异
comm -23 <(sort http_hosts.txt) <(sort sni_hosts.txt)

# Python实现SNI/Host比对检测
python3 -c "
import json
from collections import defaultdict

pairs = defaultdict(set)
# 模拟流量解析
traffic = [
    {'sni': 'www.cloudflare.com', 'host': 'evil-c2.attacker.com'},
    {'sni': 'www.cloudflare.com', 'host': 'www.cloudflare.com'},
    {'sni': 'www.akamai.com', 'host': 'c2-panel.evil.net'},
]
for t in traffic:
    pairs[t['sni']].add(t['host'])

for sni, hosts in pairs.items():
    if len(hosts) > 1:
        print(f'Domain Fronting detected on {sni}:')
        for h in hosts:
            print(f'  -> {h}')
"

import socket
import time
from collections import defaultdict

def detect_fast_flux(target_domain, samples=10, interval=30):
    ip_history = defaultdict(int)
    ns_history = defaultdict(int)

    for i in range(samples):
        try:
            answers = socket.getaddrinfo(target_domain, 80, socket.AF_INET)
            current_ips = set()
            for ans in answers:
                ip = ans[4][0]
                current_ips.add(ip)
                ip_history[ip] += 1
            print(f'Sample {i+1}: IPs={current_ips}')

            if i % 3 == 0:
                import dns.resolver
                ns_records = dns.resolver.resolve(target_domain, 'NS')
                for ns in ns_records:
                    ns_history[str(ns)] += 1

            time.sleep(interval)
        except Exception as e:
            print(f'Sample {i+1} failed: {e}')

    flux_score = {
        'ip_variety': len(ip_history) / max(sum(ip_history.values()), 1),
        'ns_variety': len(ns_history) / max(sum(ns_history.values()), 1),
        'is_fast_flux': len(ip_history) > 5 and max(ip_history.values()) < samples
    }
    return flux_score

if __name__ == '__main__':
    result = detect_fast_flux('suspicious-domain.example.com')
    print(f'Fast Flux Score: {result}')

title: DNS Tunneling Detection via High Query Rate and High Entropy
id: d9e4f5a6-b7c8-4d9e-0f1a-2b3c4d5e6f7a
status: experimental
description: Detects potential DNS tunneling activity characterized by high query rate from a single source combined with high-entropy subdomain queries
author: BlueTeam Analyst
date: 2026-07-08
tags:
  - attack.t1572
  - attack.ta0011
logsource:
  category: dns
  product: windows
  service: dns-server
detection:
  selection_high_rate:
    EventID: 264
    SourceIp: 
      - '10.0.0.0/8'
      - '172.16.0.0/12'
      - '192.168.0.0/16'
  selection_high_entropy:
    QueryName|re: '[a-z2-7]{30,}\.[a-z0-9]+\.[a-z]+'
    QueryType:
      - 'TXT'
      - 'MX'
      - 'NULL'
  timeframe: 5m
  condition: selection_high_rate and selection_high_entropy | count() > 50 by SourceIp
falsepositives:
  - CDN health check services
  - DNS zone transfer tools
  - Security scanners with TXT record verification
level: high

title: DGA Domain NXDOMAIN Burst Detection
id: e0f1a2b3-c4d5-4e6f-7a8b-9c0d1e2f3a4b
status: experimental
description: Detects burst of NXDOMAIN responses from a single client, indicative of DGA domain generation and resolution attempts
author: BlueTeam Analyst
date: 2026-07-08
tags:
  - attack.t1568.002
  - attack.ta0011
logsource:
  category: dns
  product: linux
  service: bind
detection:
  selection:
    query_type: 'A'
    return_code: 'NXDOMAIN'
  timeframe: 10m
  condition: selection | count() > 200 by client_ip
falsepositives:
  - Web crawlers with malformed URLs
  - Misconfigured applications
  - Internal scanning tools
level: medium

#!/bin/bash
# DNS日志批量分析脚本
# 检测DNS隧道、DGA、异常查询模式

DNS_LOG="${1:-/var/log/named/named.log}"
ANALYSIS_DIR="dns_analysis_$(date +%Y%m%d)"
mkdir -p "$ANALYSIS_DIR"

echo "[*] Starting DNS log analysis on $DNS_LOG"
echo "[*] Results will be saved to $ANALYSIS_DIR/"

# 1. 提取所有查询域名并统计频次
echo "[1/6] Extracting query domains..."
grep -oP 'query: \K[^ ]+' "$DNS_LOG" | \
  sed 's/\.$//' | sort | uniq -c | sort -rn > \
  "$ANALYSIS_DIR/query_frequency.txt"
echo "  Total unique domains: $(wc -l < $ANALYSIS_DIR/query_frequency.txt)"

# 2. 检测长子域名
echo "[2/6] Detecting long subdomains..."
awk '{print $2}' "$ANALYSIS_DIR/query_frequency.txt" | \
  awk -F'.' '{
    if(length($1) > 40) print length($1), $0
  }' | sort -rn | head -50 > "$ANALYSIS_DIR/long_subdomains.txt"
echo "  Domains with >40 char subdomain: $(wc -l < $ANALYSIS_DIR/long_subdomains.txt)"

# 3. 检测NXDOMAIN错误
echo "[3/6] Detecting NXDOMAIN errors..."
grep -c "NXDOMAIN" "$DNS_LOG" > "$ANALYSIS_DIR/nxdomain_count.txt"
echo "  NXDOMAIN count: $(cat $ANALYSIS_DIR/nxdomain_count.txt)"

# 4. 提取高频查询的源IP
echo "[4/6] Extracting top query sources..."
grep -oP 'client \K[0-9.]+' "$DNS_LOG" | \
  sort | uniq -c | sort -rn | head -20 > \
  "$ANALYSIS_DIR/top_sources.txt"
echo "  Top sources saved"

# 5. 检测高熵域名（DGA候选）
echo "[5/6] Detecting high-entropy domains..."
while read -r count domain; do
  entropy=$(echo "$domain" | grep -oP '^[a-z0-9]+' | \
    python3 -c "
import sys, math
data = sys.stdin.read().strip()
if not data:
    exit(0)
entropy = 0
for c in range(256):
    p = data.count(chr(c)) / len(data)
    if p > 0:
        entropy -= p * math.log2(p)
print(f'{entropy:.2f}')
")
  if [ -n "$entropy" ] && [ "$(echo "$entropy > 4.0" | bc -l 2>/dev/null)" = "1" ]; then
    echo "$entropy $count $domain" >> "$ANALYSIS_DIR/high_entropy_domains.txt"
  fi
done < "$ANALYSIS_DIR/query_frequency.txt"

sort -rn "$ANALYSIS_DIR/high_entropy_domains.txt" -o "$ANALYSIS_DIR/high_entropy_domains.txt"
echo "  High-entropy domains: $(wc -l < $ANALYSIS_DIR/high_entropy_domains.txt)"

# 6. 汇总报告
echo "[6/6] Generating summary report..."
{
  echo "=== DNS Log Analysis Report ==="
  echo "Analysis date: $(date)"
  echo "Log file: $DNS_LOG"
  echo ""
  echo "--- Top 10 Most Queried Domains ---"
  head -10 "$ANALYSIS_DIR/query_frequency.txt"
  echo ""
  echo "--- Top 5 Long Subdomains ---"
  head -5 "$ANALYSIS_DIR/long_subdomains.txt"
  echo ""
  echo "--- Top 5 High-Entropy Domains (DGA candidate) ---"
  head -5 "$ANALYSIS_DIR/high_entropy_domains.txt"
} > "$ANALYSIS_DIR/summary_report.txt"

echo "[*] Analysis complete. Report: $ANALYSIS_DIR/summary_report.txt"

#!/usr/bin/env python3
import dpkt
import socket
import base64
import struct
import sys
from collections import defaultdict

class DNSTunnelExtractor:
    def __init__(self, pcap_file):
        self.pcap_file = pcap_file
        self.tunnel_candidates = defaultdict(list)
        self.extracted_data = {}
        self.suspicious_ip_ranges = [
            ('10.0.0.0', '10.255.255.255'),
            ('172.16.0.0', '172.31.255.255'),
            ('192.168.0.0', '192.168.255.255')
        ]

    def ip_to_int(self, ip_str):
        parts = ip_str.split('.')
        return (int(parts[0]) << 24) + (int(parts[1]) << 16) + \
               (int(parts[2]) << 8) + int(parts[3])

    def is_private_ip(self, ip_str):
        ip_int = self.ip_to_int(ip_str)
        for start, end in self.suspicious_ip_ranges:
            start_int = self.ip_to_int(start)
            end_int = self.ip_to_int(end)
            if start_int <= ip_int <= end_int:
                return True
        return False

    def parse_dns_packets(self):
        with open(self.pcap_file, 'rb') as f:
            pcap = dpkt.pcap.Reader(f)
            for ts, buf in pcap:
                try:
                    eth = dpkt.ethernet.Ethernet(buf)
                    ip = eth.data
                    if isinstance(ip, dpkt.ip.IP) and ip.p == dpkt.ip.IP_PROTO_UDP:
                        udp = ip.data
                        if udp.dport == 53 or udp.sport == 53:
                            dns = dpkt.dns.DNS(udp.data)
                            if len(dns.qd) > 0:
                                q = dns.qd[0]
                                qname = q.name.decode('utf-8', errors='ignore')
                                qtype = q.type
                                src_ip = socket.inet_ntoa(ip.src)
                                dst_ip = socket.inet_ntoa(ip.dst)
                                if self.is_private_ip(src_ip):
                                    labels = qname.rstrip('.').split('.')
                                    if len(labels) >= 3 and len(labels[0]) > 30:
                                        tunnel_domain = '.'.join(labels[-3:])
                                        encoded_part = labels[0]
                                        self.tunnel_candidates[src_ip].append({
                                            'timestamp': ts,
                                            'domain': tunnel_domain,
                                            'encoded': encoded_part,
                                            'qtype': qtype,
                                            'dst_ip': dst_ip
                                        })
                except Exception:
                    continue

    def decode_tunnel_data(self):
        for src_ip, entries in self.tunnel_candidates.items():
            encoded_data = ''
            domain_group = defaultdict(list)
            for entry in entries:
                domain_group[entry['domain']].append(entry)
            for domain, group in domain_group.items():
                group.sort(key=lambda x: x['timestamp'])
                for entry in group:
                    try:
                        raw = entry['encoded'].replace('-', '/').replace('_', '+')
                        padding = 4 - len(raw) % 4
                        if padding != 4:
                            raw += '=' * padding
                        decoded = base64.b64decode(raw)
                        encoded_data += decoded.decode('utf-8', errors='replace')
                    except Exception:
                        continue
                if encoded_data:
                    self.extracted_data[domain] = {
                        'src_ip': src_ip,
                        'packet_count': len(group),
                        'total_bytes': len(encoded_data),
                        'decoded_preview': encoded_data[:200]
                    }

    def generate_report(self):
        print(f'[*] DNS Tunnel Analysis Report')
        print(f'[*] PCAP: {self.pcap_file}')
        print(f'[*] Tunnel candidates: {sum(len(v) for v in self.tunnel_candidates.values())}')
        print(f'[*] Unique source IPs: {len(self.tunnel_candidates)}')
        print()
        for domain, info in self.extracted_data.items():
            print(f'[!] Tunnel detected: {domain}')
            print(f'    Source: {info["src_ip"]}')
            print(f'    Packets: {info["packet_count"]}')
            print(f'    Data: {info["total_bytes"]} bytes')
            print(f'    Preview: {info["decoded_preview"][:80]}...')
            print()

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print(f'Usage: {sys.argv[0]} <pcap_file>')
        sys.exit(1)
    extractor = DNSTunnelExtractor(sys.argv[1])
    extractor.parse_dns_packets()
    extractor.decode_tunnel_data()
    extractor.generate_report()

域名: avsvmcloud[.]com（DGA C2）
域名: digitalcollege[.]org
域名: virtualnewspaper[.]net
IP: 13.59.140.44
IP: 54.193.121.176
签名证书: SolarWinds Worldwide, LLC

# TrickBot DGA 种子示例（2018年变种）
trickbot_dga_seeds = [
    'alltheplacesyoulike',
    'greatmistakesaremade',
    'futureislookinggood',
    'thistimereallyworks',
    'betterlatethanever',
]