detecting-dns-exfiltration-with-dns-query-analysis

DNS クエリ分析による DNS 持ち出しの検知

概要

DNS 持ち出しは、ドメインネームシステムを隠れたチャネルとして悪用し、侵害されたネットワークからデータを抽出します。攻撃者は、盗み出したデータを DNS クエリ名（サブドメイン）や DNS 応答レコード（TXT、CNAME、NULL）にエンコードすることで、通常は制限なく許可されている DNS トラフィックを対象とする従来のセキュリティ制御を回避します。iodine、dnscat2、dns2tcp などのツールを使うと、DNS 経由で TCP の完全なトンネリングが可能になります。検知には、クエリ長の過剰さ、サブドメイン文字列の高いエントロピー、単一ドメインへの異常なクエリ量、サイズの大きすぎる TXT レコード応答などの異常を見つけるため、DNS クエリパターンを分析する必要があります。このスキルでは、パッシブ DNS 分析、統計的手法、機械学習アプローチを活用して、包括的な DNS 持ち出し検知機能を構築する方法を解説します。

使用する場面

• DNS クエリ分析による DNS 持ち出しの検知が必要なセキュリティインシデントを調査するとき
• この領域の検知ルールや脅威ハンティング用クエリを構築するとき
• SOC アナリストがこの種の分析に構造化された手順を必要とするとき
• 関連する攻撃手法に対するセキュリティ監視の網羅性を検証するとき

前提条件

• DNS クエリログへのアクセス（パッシブ DNS キャプチャ、DNS サーバーログ、または PCAP）
• DNS トラフィックのキャプチャに使用する Zeek、Suricata、または tcpdump
• scipy、numpy、pandas、scikit-learn を備えた Python 3.8 以降
• アラートの相関分析に使用する SIEM プラットフォーム
• 対象環境における通常の DNS トラフィックパターンのベースライン

主な概念

DNS トンネリングの仕組み

DNS 持ち出しでは、DNS メッセージのさまざまな部分にデータをエンコードします。

アウトバウンド（クエリベースの持ち出し）：

Encoded data as subdomain labels:
dGhlIHNlY3JldCBkYXRh.exfil.attacker.com
[base64-encoded data].[tunnel domain]

Query types used: A, AAAA, CNAME, MX, TXT, NULL

インバウンド（応答ベースのコマンドチャネル）：

TXT records carry encoded commands/data in responses
CNAME records chain encoded data through multiple labels
NULL records carry arbitrary binary data

検知指標

指標	通常の DNS	DNS トンネリング
サブドメインの長さ	5～20 文字	40～253 文字
ラベル数	2～4 個	5～10 個以上
シャノンエントロピー	2.5～3.5 ビット	4.0～5.5 ビット
クエリ量（ドメインあたり）	変動あり	毎分 100～1000 件
TXT 応答サイズ	100 バイト未満	200～4000 バイト以上
一意なサブドメイン	少ない	非常に多い
クエリタイプの分布	主に A/AAAA	TXT、NULL、CNAME が多い

主なトンネリングツール

ツール	プロトコル	エンコード方式	検知の難易度
iodine	IP-over-DNS	Base32/Base64/Raw	中
dnscat2	TCP-over-DNS	Hex encoding	中
dns2tcp	TCP-over-DNS	Base64	中
DNSExfiltrator	Custom	Base64	低
Cobalt Strike DNS	C2 over DNS	Custom encoding	高

ワークフロー

ステップ 1：DNS トラフィックをキャプチャする

Zeek を使用する場合：

# Live capture
zeek -i eth0 -C base/protocols/dns

# Offline PCAP analysis
zeek -r traffic.pcap base/protocols/dns

# Output: dns.log with query, qtype, answers, TTL

tcpdump を使用する場合：

# Capture all DNS traffic
tcpdump -i eth0 -w dns_capture.pcap port 53

# Capture with size filter (large DNS packets)
tcpdump -i eth0 -w large_dns.pcap 'port 53 and greater 512'

Suricata を使用する場合：

# In suricata.yaml, enable DNS logging
outputs:
  - eve-log:
      types:
        - dns:
            query: yes
            answer: yes
            formats: [detailed]

ステップ 2：クエリの特性を分析する

DNS 持ち出しを検知する Python スクリプト：

#!/usr/bin/env python3
"""DNS Exfiltration Detector - Analyzes DNS logs for tunneling indicators."""

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:
    """Calculate Shannon entropy of a string."""
    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:
    """Extract subdomain portion from FQDN."""
    parts = query.rstrip('.').split('.')
    if len(parts) > 2:
        return '.'.join(parts[:-2])
    return ''


def get_base_domain(query: str) -> str:
    """Extract registered domain from FQDN."""
    parts = query.rstrip('.').split('.')
    if len(parts) >= 2:
        return '.'.join(parts[-2:])
    return query


def is_base64_like(s: str) -> bool:
    """Check if string resembles base64 encoding."""
    b64_chars = set('ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=')
    if len(s) < 10:
        return False
    char_ratio = sum(1 for c in s if c in b64_chars) / len(s)
    return char_ratio > 0.9 and calculate_entropy(s) > 4.0


def is_hex_encoded(s: str) -> bool:
    """Check if string appears hex-encoded."""
    hex_chars = set('0123456789abcdefABCDEF')
    if len(s) < 16:
        return False
    clean = s.replace('.', '').replace('-', '')
    return all(c in hex_chars for c in clean) and len(clean) % 2 == 0


class DNSExfiltrationDetector:
    def __init__(self):
        self.domain_stats = defaultdict(lambda: {
            'query_count': 0,
            'unique_subdomains': set(),
            'total_subdomain_length': 0,
            'entropy_sum': 0.0,
            'query_types': defaultdict(int),
            'source_ips': set(),
            'first_seen': None,
            'last_seen': None,
            'txt_response_sizes': [],
        })

        # Detection thresholds
        self.thresholds = {
            'min_query_count': 50,
            'min_unique_subdomains': 30,
            'avg_subdomain_length': 30,
            'avg_entropy': 3.8,
            'unique_ratio': 0.7,
            'txt_query_ratio': 0.3,
            'max_label_length': 63,
            'max_subdomain_labels': 5,
        }

    def process_query(self, timestamp, src_ip, query, qtype, response_size=0):
        """Process a single DNS query and update statistics."""
        base_domain = get_base_domain(query)
        subdomain = extract_subdomain(query)

        stats = self.domain_stats[base_domain]
        stats['query_count'] += 1
        stats['unique_subdomains'].add(subdomain)
        stats['total_subdomain_length'] += len(subdomain)
        stats['entropy_sum'] += calculate_entropy(subdomain)
        stats['query_types'][qtype] += 1
        stats['source_ips'].add(src_ip)

        if stats['first_seen'] is None:
            stats['first_seen'] = timestamp
        stats['last_seen'] = timestamp

        if qtype in ('TXT', 'NULL') and response_size > 0:
            stats['txt_response_sizes'].append(response_size)

    def analyze(self):
        """Analyze accumulated statistics and return suspicious domains."""
        alerts = []

        for domain, stats in self.domain_stats.items():
            if stats['query_count'] < self.thresholds['min_query_count']:
                continue

            unique_count = len(stats['unique_subdomains'])
            avg_length = stats['total_subdomain_length'] / stats['query_count']
            avg_entropy = stats['entropy_sum'] / stats['query_count']
            unique_ratio = unique_count / stats['query_count']

            txt_queries = stats['query_types'].get('TXT', 0) + stats['query_types'].get('NULL', 0)
            txt_ratio = txt_queries / stats['query_count']

            score = 0
            indicators = []

            if avg_length > self.thresholds['avg_subdomain_length']:
                score += 25
                indicators.append(f"high_avg_subdomain_length={avg_length:.1f}")

            if avg_entropy > self.thresholds['avg_entropy']:
                score += 25
                indicators.append(f"high_entropy={avg_entropy:.2f}")

            if unique_ratio > self.thresholds['unique_ratio']:
                score += 20
                indicators.append(f"high_unique_ratio={unique_ratio:.2f}")

            if txt_ratio > self.thresholds['txt_query_ratio']:
                score += 15
                indicators.append(f"high_txt_ratio={txt_ratio:.2f}")

            if unique_count > self.thresholds['min_unique_subdomains']:
                score += 15
                indicators.append(f"unique_subdomains={unique_count}")

            # Check for encoding patterns
            encoded_count = sum(
                1 for sd in list(stats['unique_subdomains'])[:100]
                if is_base64_like(sd) or is_hex_encoded(sd)
            )
            if encoded_count > 20:
                score += 20
                indicators.append(f"encoded_subdomains={encoded_count}")

            if score >= 50:
                duration = (stats['last_seen'] - stats['first_seen']).total_seconds() if stats['first_seen'] and stats['last_seen'] else 0
                alerts.append({
                    'domain': domain,
                    'score': min(score, 100),
                    'query_count': stats['query_count'],
                    'unique_subdomains': unique_count,
                    'avg_subdomain_length': round(avg_length, 1),
                    'avg_entropy': round(avg_entropy, 2),
                    'unique_ratio': round(unique_ratio, 2),
                    'txt_ratio': round(txt_ratio, 2),
                    'source_ips': list(stats['source_ips']),
                    'duration_seconds': duration,
                    'indicators': indicators,
                })

        return sorted(alerts, key=lambda x: x['score'], reverse=True)

    def process_zeek_dns_log(self, log_path):
        """Process Zeek dns.log file."""
        with open(log_path, 'r') as f:
            for line in f:
                if line.startswith('#'):
                    continue
                fields = line.strip().split('\t')
                if len(fields) < 22:
                    continue
                try:
                    ts = datetime.fromtimestamp(float(fields[0]))
                    src_ip = fields[2]
                    query = fields[9]
                    qtype = fields[11]
                    self.process_query(ts, src_ip, query, qtype)
                except (ValueError, IndexError):
                    continue

    def process_eve_json(self, log_path):
        """Process Suricata EVE JSON DNS log."""
        with open(log_path, 'r') as f:
            for line in f:
                try:
                    event = json.loads(line)
                    if event.get('event_type') != 'dns':
                        continue
                    dns = event.get('dns', {})
                    ts = datetime.fromisoformat(event['timestamp'].replace('Z', '+00:00'))
                    src_ip = event.get('src_ip', '')
                    query = dns.get('rrname', '')
                    qtype = dns.get('rrtype', '')
                    self.process_query(ts, src_ip, query, qtype)
                except (json.JSONDecodeError, KeyError, ValueError):
                    continue


def main():
    detector = DNSExfiltrationDetector()

    log_file = sys.argv[1] if len(sys.argv) > 1 else '/opt/zeek/logs/current/dns.log'

    if log_file.endswith('.json'):
        detector.process_eve_json(log_file)
    else:
        detector.process_zeek_dns_log(log_file)

    alerts = detector.analyze()

    if alerts:
        print(f"\n{'='*80}")
        print(f"DNS EXFILTRATION DETECTION RESULTS - {len(alerts)} suspicious domains found")
        print(f"{'='*80}\n")

        for alert in alerts:
            severity = "CRITICAL" if alert['score'] >= 80 else "HIGH" if alert['score'] >= 60 else "MEDIUM"
            print(f"[{severity}] Domain: {alert['domain']}")
            print(f"  Score: {alert['score']}/100")
            print(f"  Queries: {alert['query_count']}, Unique Subdomains: {alert['unique_subdomains']}")
            print(f"  Avg Subdomain Length: {alert['avg_subdomain_length']}, Avg Entropy: {alert['avg_entropy']}")
            print(f"  Source IPs: {', '.join(alert['source_ips'][:5])}")
            print(f"  Indicators: {', '.join(alert['indicators'])}")
            print()
    else:
        print("No DNS exfiltration indicators detected.")


if __name__ == '__main__':
    main()

ステップ 3：DNS 持ち出し向けの Suricata ルールを適用する

# Detect long DNS queries (potential tunneling)
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;)

# Detect high-entropy DNS subdomain
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;)

# Detect large TXT record responses
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;)

# Detect NULL record queries (used by 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;)

# Detect dnscat2 traffic pattern
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 検知ルール

DNS 持ち出しを検知する Splunk SPL クエリ：

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

対応アクション

1. トンネルのドメインをブロックする（DNS リゾルバーとファイアウォールの両方で）
2. 送信元ホストをネットワークから隔離する（フォレンジック調査のため）
3. 証拠保全のため、DNS トラフィックの完全な PCAP を取得する
4. キャプチャした DNS クエリをデコードして、持ち出されたデータを特定する
5. 侵害されたホストに永続化の仕組みがないか確認する
6. 特定した C2 ドメインとインフラストラクチャをブロックリストに追加する

ベストプラクティス

• DNS ロギング - リゾルバーとネットワークレベルで、すべての DNS クエリと応答のログを有効にします
• 内部 DNS のみを使用 - すべての DNS トラフィックを内部リゾルバー経由にし、外部 DNS への直接アクセス（ポート 53）をブロックします
• レスポンスポリシーゾーン - RPZ フィードを導入して、既知のトンネリングドメインをブロックします
• まずベースラインを確立 - 検知しきい値を設定する前に、通常の DNS クエリパターンを確立します
• TXT レコードを監視 - TXT および NULL レコードのクエリに特に注意します
• 暗号化 DNS に留意 - DNS の検査を回避する可能性がある DoH/DoT の使用を監視します

参考資料

• Splunk：ディープラーニングによる DNS データ流出の検知
• Akamai：DNS データ流出
• SANS：DNS トンネリングの検知
• Fidelis：DNS トンネリングの検知