#!/usr/bin/env python3 """ pe_info.py - one-shot static PE triage summary. Runs entirely on the host against an exported/copied sample file. Never executes the sample. Prints a plain-text triage report (or JSON with --json) covering headers, sections, packing heuristics, imports/exports, signature, resources, overlay, PDB path, version info, TLS callbacks and rich header. Usage: python3 pe_info.py [--json] Requires: pefile (use an approved/pinned installation; do not auto-install) """ import sys import os import hashlib import math import json import argparse from datetime import datetime, timezone try: import pefile except ImportError: print("pefile is required; use an approved/pinned installation or fallback tooling", file=sys.stderr) sys.exit(2) PACKER_SECTION_NAMES = { "UPX0", "UPX1", ".THEMIDA", ".VMP0", ".VMP1", ".ASPACK", ".MPRESS1", ".ENIGMA1", ".PETITE", ".NSP0", ".UPACK", } SUBSYSTEM_NAMES = { 1: "NATIVE", 2: "GUI", 3: "CUI", 5: "OS2 CUI", 7: "POSIX CUI", 9: "WINDOWS CE GUI", 10: "EFI APPLICATION", 11: "EFI BOOT SERVICE DRIVER", 12: "EFI RUNTIME DRIVER", 13: "EFI ROM", 14: "XBOX", 16: "WINDOWS BOOT APPLICATION", } MACHINE_NAMES = { 0x14c: "x86", 0x8664: "x64", 0xaa64: "ARM64", 0x1c0: "ARM", 0x1c4: "ARM Thumb-2", 0x200: "IA64", } SUSPICIOUS_CATEGORIES = { "injection": ["CreateRemoteThread", "WriteProcessMemory", "VirtualAllocEx", "OpenProcess", "NtUnmapViewOfSection", "SetThreadContext", "ResumeThread", "QueueUserAPC", "NtCreateThreadEx", "RtlCreateUserThread"], "keylogging": ["SetWindowsHookEx", "GetAsyncKeyState", "GetKeyState", "GetForegroundWindow"], "network": ["InternetOpen", "InternetOpenUrl", "InternetReadFile", "HttpOpenRequest", "HttpSendRequest", "URLDownloadToFile", "WSAStartup", "socket", "connect", "send", "recv", "WinHttpOpen", "WinHttpConnect", "DnsQuery", "getaddrinfo", "gethostbyname"], "persistence": ["RegSetValueEx", "RegCreateKeyEx", "CreateService", "StartService", "ChangeServiceConfig", "CoCreateInstance"], "anti-analysis": ["IsDebuggerPresent", "CheckRemoteDebuggerPresent", "NtQueryInformationProcess", "OutputDebugString", "GetTickCount", "QueryPerformanceCounter", "FindWindow", "GetLastInputInfo", "GetCursorPos", "GlobalMemoryStatusEx", "GetSystemInfo", "EnumProcesses", "CreateToolhelp32Snapshot", "Process32First", "Process32Next"], "evasion": ["UpdateProcThreadAttribute", "VirtualProtect", "EtwEventWrite", "AmsiScanBuffer", "AmsiScanString", "NtTraceEvent", "AddVectoredExceptionHandler"], "crypto": ["CryptEncrypt", "CryptDecrypt", "CryptAcquireContext", "CryptGenKey", "CryptHashData", "BCryptEncrypt", "BCryptDecrypt", "CryptImportKey"], "file-ops": ["CreateFile", "WriteFile", "DeleteFile", "MoveFile", "CopyFile", "FindFirstFile", "FindNextFile", "SetFileAttributes"], "execution": ["CreateProcess", "ShellExecute", "WinExec", "system", "CreateProcessAsUser", "CreateProcessWithLogon"], "callback-exec": ["EnumChildWindows", "EnumDesktopWindows", "CreateTimerQueueTimer", "EnumSystemLocales", "EnumTimeFormatsEx", "SetTimer", "TpAllocWork"], "dynamic-loading": ["LoadLibrary", "LoadLibraryEx", "GetProcAddress", "LdrLoadDll"], } INJECTION_APIS = {n.lower() for n in SUSPICIOUS_CATEGORIES["injection"]} DYNLOAD_APIS = {n.lower() for n in SUSPICIOUS_CATEGORIES["dynamic-loading"]} def norm_api_name(name: str) -> str: """Normalize a WinAPI import name for matching: strip stdcall decoration (@N / leading _), strip a trailing Ansi/Wide 'A'/'W' suffix, lowercase.""" n = name.strip() if n.startswith("_"): n = n[1:] if "@" in n: n = n.split("@")[0] if len(n) > 1 and n[-1] in ("A", "W") and n[-2].isalpha(): n = n[:-1] return n.lower() CATEGORY_LOOKUP = {} for _cat, _names in SUSPICIOUS_CATEGORIES.items(): for _n in _names: CATEGORY_LOOKUP.setdefault(norm_api_name(_n), []).append(_cat) def shannon_entropy(data: bytes) -> float: if not data: return 0.0 freq = [0] * 256 for b in data: freq[b] += 1 length = len(data) entropy = 0.0 for f in freq: if f: p = f / length entropy -= p * math.log2(p) return entropy def hash_file(path: str): md5, sha1, sha256 = hashlib.md5(), hashlib.sha1(), hashlib.sha256() with open(path, "rb") as f: for chunk in iter(lambda: f.read(1 << 20), b""): md5.update(chunk) sha1.update(chunk) sha256.update(chunk) return md5.hexdigest(), sha1.hexdigest(), sha256.hexdigest() def decode(b): if b is None: return None if isinstance(b, bytes): return b.rstrip(b"\x00").decode(errors="replace") return str(b) class Report: """Collects text lines + a parallel structured dict for --json.""" def __init__(self): self.lines = [] self.data = {} def h(self, title): self.lines.append("") self.lines.append(f"=== {title} ===") def p(self, text=""): self.lines.append(text) def fail(self, section, err): self.lines.append(f"section failed: {err}") self.data.setdefault("errors", {})[section] = str(err) def section_file_info(pe, path, rpt: Report): rpt.h("File") size = os.path.getsize(path) md5, sha1, sha256 = hash_file(path) with open(path, "rb") as f: data = f.read() entropy = shannon_entropy(data) try: imphash = pe.get_imphash() except Exception as e: imphash = f"n/a ({e})" magic = pe.OPTIONAL_HEADER.Magic pe_type = "PE32+" if magic == 0x20B else "PE32" if magic == 0x10B else f"unknown (0x{magic:x})" machine_val = pe.FILE_HEADER.Machine machine = MACHINE_NAMES.get(machine_val, f"0x{machine_val:x}") subsys_val = pe.OPTIONAL_HEADER.Subsystem subsystem = SUBSYSTEM_NAMES.get(subsys_val, f"0x{subsys_val:x}") is_dll = pe.is_dll() kind = "DLL" if is_dll else "EXE" com_dir = pe.OPTIONAL_HEADER.DATA_DIRECTORY[ pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_COM_DESCRIPTOR"] ] is_dotnet = com_dir.VirtualAddress != 0 ts = pe.FILE_HEADER.TimeDateStamp ts_flag = "" try: ts_dt = datetime.fromtimestamp(ts, tz=timezone.utc) now = datetime.now(tz=timezone.utc) if ts_dt > now: ts_flag = " [FLAG: timestamp is in the future]" if is_dotnet: ts_flag += " (normal for .NET deterministic builds: field holds a hash, not a date)" elif ts_dt.year < 2000: ts_flag = " [FLAG: timestamp before year 2000]" ts_str = ts_dt.isoformat() except (OverflowError, OSError, ValueError): ts_str = f"invalid raw value 0x{ts:x}" ts_flag = " [FLAG: unparsable timestamp]" rpt.p(f"Path: {path}") rpt.p(f"Size: {size} bytes") rpt.p(f"MD5: {md5}") rpt.p(f"SHA1: {sha1}") rpt.p(f"SHA256: {sha256}") rpt.p(f"Entropy: {entropy:.2f} (whole file)") rpt.p(f"Imphash: {imphash}") rpt.p(f"Format: {pe_type}, machine={machine}, subsystem={subsystem}, {kind}") rpt.p(f".NET: {'YES - route to specialized-file-analyzer' if is_dotnet else 'no'}") rpt.p(f"Compiled: {ts_str}{ts_flag}") rpt.p(f"Linker: {pe.OPTIONAL_HEADER.MajorLinkerVersion}.{pe.OPTIONAL_HEADER.MinorLinkerVersion}") rpt.data["file"] = { "path": path, "size": size, "md5": md5, "sha1": sha1, "sha256": sha256, "entropy": round(entropy, 2), "imphash": imphash, "pe_type": pe_type, "machine": machine, "subsystem": subsystem, "is_dll": is_dll, "is_dotnet": is_dotnet, "compile_time_utc": ts_str, "linker_version": f"{pe.OPTIONAL_HEADER.MajorLinkerVersion}.{pe.OPTIONAL_HEADER.MinorLinkerVersion}", } return is_dotnet def section_sections(pe, rpt: Report): rpt.h("Sections") ep = pe.OPTIONAL_HEADER.AddressOfEntryPoint ep_section = pe.get_section_by_rva(ep) ep_name = decode(ep_section.Name) if ep_section else None rows = [] header = f"{'Name':<10} {'VirtSize':>10} {'RawSize':>10} {'Entropy':>8} {'Flags':>6} Notes" rpt.p(header) rpt.p("-" * len(header)) for i, sec in enumerate(pe.sections): name = decode(sec.Name) vsize = sec.Misc_VirtualSize rsize = sec.SizeOfRawData try: ent = sec.get_entropy() except Exception: ent = 0.0 chars = sec.Characteristics flags = "" flags += "R" if chars & 0x40000000 else "-" flags += "W" if chars & 0x80000000 else "-" flags += "X" if chars & 0x20000000 else "-" notes = [] packer_hit = name.strip().upper() in PACKER_SECTION_NAMES high_entropy = ent > 7.0 if packer_hit or high_entropy: notes.append("!") if name == ep_name and ep_section is not None: notes.append("[ENTRY POINT]") rows.append({ "name": name, "virtual_size": vsize, "raw_size": rsize, "entropy": round(ent, 2), "flags": flags, "packer_name_match": packer_hit, "high_entropy": high_entropy, }) rpt.p(f"{name:<10} {vsize:>10} {rsize:>10} {ent:>8.2f} {flags:>6} {' '.join(notes)}") if ep_section is None: rpt.p("WARNING: entry point RVA does not resolve to any section") else: if ep_name not in (".text", "CODE"): rpt.p(f"WARNING: entry point is in section '{ep_name}', not .text/CODE") if pe.sections and ep_section is pe.sections[-1]: rpt.p(f"WARNING: entry point is in the LAST section ('{ep_name}') - common packer stub pattern") rpt.data["sections"] = rows return rows def section_packing_verdict(pe, rpt: Report, overall_entropy, sections, total_imports, kernel32_only_loader_pattern, is_dotnet=False): rpt.h("Packing verdict") reasons = [] if overall_entropy > 7.0: reasons.append(f"overall file entropy {overall_entropy:.2f} > 7.0") exec_high = [s for s in sections if "X" in s["flags"] and s["entropy"] > 6.8] if exec_high: names = ", ".join(s["name"] for s in exec_high) reasons.append(f"executable section entropy > 6.8 in: {names}") if is_dotnet: rpt.p("Note: .NET assembly - import-table heuristics skipped (managed code resolves calls via CLR metadata).") else: if total_imports < 10: reasons.append(f"only {total_imports} imported functions total") if kernel32_only_loader_pattern: reasons.append("imports are only from KERNEL32 with a LoadLibrary/GetProcAddress/VirtualAlloc/VirtualProtect pattern") if reasons: rpt.p("Likely packed. Reasons:") for r in reasons: rpt.p(f" - {r}") else: rpt.p("No strong packing indicators.") rpt.data["packing_verdict"] = {"likely_packed": bool(reasons), "reasons": reasons} def section_imports(pe, rpt: Report): rpt.h("Imports") total = 0 category_hits = {} dll_names = set() kernel32_functions = set() all_dll_only_kernel32 = True if hasattr(pe, "DIRECTORY_ENTRY_IMPORT"): for entry in pe.DIRECTORY_ENTRY_IMPORT: dll = decode(entry.dll) or "?" dll_names.add(dll.lower()) if dll.lower() != "kernel32.dll": all_dll_only_kernel32 = False names = [] for imp in entry.imports: total += 1 fname = decode(imp.name) if imp.name else f"ordinal#{imp.ordinal}" names.append(fname) if dll.lower() == "kernel32.dll" and imp.name: kernel32_functions.add(norm_api_name(fname)) if imp.name: cats = CATEGORY_LOOKUP.get(norm_api_name(fname)) if cats: for c in cats: category_hits.setdefault(c, set()).add(fname) rpt.p(f"{dll} ({len(names)} functions):") rpt.p(" " + ", ".join(names)) else: rpt.p("No import table.") rpt.p("") rpt.p("Suspicious imports by category:") if category_hits: for cat in sorted(category_hits): names = ", ".join(sorted(category_hits[cat])) rpt.p(f" {cat}: {names}") else: rpt.p(" none matched") loader_pattern = {"loadlibrary", "getprocaddress", "virtualalloc", "virtualprotect"} kernel32_loader_pattern = ( dll_names == {"kernel32.dll"} and loader_pattern.issubset(kernel32_functions) ) # delay-load imports rpt.p("") rpt.p("Delay-load imports:") if hasattr(pe, "DIRECTORY_ENTRY_DELAY_IMPORT") and pe.DIRECTORY_ENTRY_DELAY_IMPORT: for entry in pe.DIRECTORY_ENTRY_DELAY_IMPORT: dll = decode(entry.dll) or "?" names = [decode(i.name) if i.name else f"ordinal#{i.ordinal}" for i in entry.imports] rpt.p(f" {dll} ({len(names)}): {', '.join(names)}") else: rpt.p(" none") # exports rpt.p("") rpt.p("Exports:") exports = [] if hasattr(pe, "DIRECTORY_ENTRY_EXPORT"): for sym in pe.DIRECTORY_ENTRY_EXPORT.symbols[:50]: exports.append(decode(sym.name) if sym.name else f"ordinal#{sym.ordinal}") rpt.p(" " + ", ".join(exports) if exports else " (export directory present, no symbols)") total_exports = len(pe.DIRECTORY_ENTRY_EXPORT.symbols) if total_exports > 50: rpt.p(f" ... ({total_exports} total, showing first 50)") else: rpt.p(" none") rpt.data["imports"] = { "total_functions": total, "categories": {k: sorted(v) for k, v in category_hits.items()}, "exports_count": len(exports), } return total, kernel32_loader_pattern def section_signature(pe, rpt: Report): rpt.h("Signature") sec_dir = pe.OPTIONAL_HEADER.DATA_DIRECTORY[ pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_SECURITY"] ] present = sec_dir.VirtualAddress != 0 and sec_dir.Size != 0 rpt.p(f"Authenticode directory present: {'yes' if present else 'no'} (size={sec_dir.Size})") rpt.p("Not verifying signature here - verify with `osslsigncode verify` or `signtool verify`.") rpt.data["signature"] = {"present": present, "size": sec_dir.Size} def section_resources(pe, rpt: Report): rpt.h("Resources") if not hasattr(pe, "DIRECTORY_ENTRY_RESOURCE"): rpt.p("No resource directory.") rpt.data["resources"] = {"count": 0} return leaves = [] has_manifest = False def type_name(entry): if entry.name is not None: return decode(entry.name.string) if hasattr(entry.name, "string") else str(entry.name) rid = entry.struct.Id name = None try: name = pefile.RESOURCE_TYPE.get(rid) except Exception: name = None return name or f"RT_UNKNOWN_{rid}" for res_type in pe.DIRECTORY_ENTRY_RESOURCE.entries: tname = type_name(res_type) if tname == "RT_MANIFEST": has_manifest = True if not hasattr(res_type, "directory"): continue for res_id in res_type.directory.entries: if not hasattr(res_id, "directory"): continue for res_lang in res_id.directory.entries: if not hasattr(res_lang, "data"): continue rva = res_lang.data.struct.OffsetToData size = res_lang.data.struct.Size leaves.append((tname, rva, size)) rpt.p(f"Total resource entries: {len(leaves)}") rpt.p(f"MANIFEST present: {'yes' if has_manifest else 'no'}") type_counts = {} for tname, _, _ in leaves: type_counts[tname] = type_counts.get(tname, 0) + 1 for tname, count in sorted(type_counts.items()): rpt.p(f" {tname}: {count}") rpt.p("Entries larger than 4KB:") large_found = False large_entries = [] for tname, rva, size in leaves: if size > 4096: large_found = True try: data = pe.get_data(rva, size) ent = shannon_entropy(data) except Exception as e: ent = None flag = " [FLAG: possible embedded payload]" if (ent is not None and ent > 7.0) else "" ent_str = f"{ent:.2f}" if ent is not None else "n/a" rpt.p(f" {tname}: {size} bytes, entropy={ent_str}{flag}") large_entries.append({"type": tname, "size": size, "entropy": ent}) if not large_found: rpt.p(" none") rpt.data["resources"] = { "count": len(leaves), "has_manifest": has_manifest, "type_counts": type_counts, "large_entries": large_entries, } def section_overlay(pe, rpt: Report, path): rpt.h("Overlay") try: overlay_offset = pe.get_overlay_data_start_offset() except Exception: overlay_offset = None if overlay_offset is None: rpt.p("No overlay.") rpt.data["overlay"] = {"size": 0} return file_size = os.path.getsize(path) size = file_size - overlay_offset rpt.p(f"Overlay size: {size} bytes (starts at offset 0x{overlay_offset:x})") guess = "unknown" head = b"" if size > 0: with open(path, "rb") as f: f.seek(overlay_offset) head = f.read(8) hex_head = head.hex() rpt.p(f"First 8 bytes: {hex_head}") if head[:2] == b"PK": guess = "PK zip (possible appended archive/installer payload)" elif head[:2] == b"MZ": guess = "MZ (possible embedded PE)" elif head[:6] == b"7z\xbc\xaf\x27\x1c": guess = "7z archive" elif head[:4] == b"UPX!": guess = "UPX marker" elif b"NullsoftInst" in head: guess = "NSIS installer marker" elif b"Inno Setup" in head: guess = "Inno Setup marker" else: sec_dir = pe.OPTIONAL_HEADER.DATA_DIRECTORY[ pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_SECURITY"] ] if sec_dir.VirtualAddress == overlay_offset: guess = "Authenticode certificate table (not a separate overlay payload)" rpt.p(f"Guess: {guess}") rpt.data["overlay"] = {"size": size, "offset": overlay_offset, "first_bytes": head.hex(), "guess": guess} def section_pdb(pe, rpt: Report): rpt.h("PDB") pdb_path = None if hasattr(pe, "DIRECTORY_ENTRY_DEBUG"): for dbg in pe.DIRECTORY_ENTRY_DEBUG: entry = getattr(dbg, "entry", None) if entry is not None and hasattr(entry, "PdbFileName"): pdb_path = decode(entry.PdbFileName) break rpt.p(f"PDB path: {pdb_path if pdb_path else 'none found'}") rpt.data["pdb_path"] = pdb_path def section_version_info(pe, rpt: Report, path): rpt.h("Version info") fields = {} if hasattr(pe, "FileInfo"): for fileinfo_list in pe.FileInfo: for entry in fileinfo_list: if getattr(entry, "Key", None) == b"StringFileInfo": for st in entry.StringTable: for k, v in st.entries.items(): fields[decode(k)] = decode(v) wanted = ["CompanyName", "ProductName", "OriginalFilename", "FileDescription", "FileVersion"] any_found = False for w in wanted: if w in fields: any_found = True rpt.p(f"{w}: {fields[w]}") if not any_found: rpt.p("No version info resource found.") orig = fields.get("OriginalFilename") if orig and orig.lower() != os.path.basename(path).lower(): rpt.p(f"WARNING: OriginalFilename ('{orig}') does not match actual filename ('{os.path.basename(path)}')") rpt.data["version_info"] = fields def section_tls(pe, rpt: Report): rpt.h("TLS callbacks") if not hasattr(pe, "DIRECTORY_ENTRY_TLS") or pe.DIRECTORY_ENTRY_TLS is None: rpt.p("No TLS directory.") rpt.data["tls_callbacks"] = 0 return cb_rva = pe.DIRECTORY_ENTRY_TLS.struct.AddressOfCallBacks if not cb_rva: rpt.p("TLS directory present but no callbacks.") rpt.data["tls_callbacks"] = 0 return image_base = pe.OPTIONAL_HEADER.ImageBase is_64 = pe.OPTIONAL_HEADER.Magic == 0x20B read_ptr = pe.get_qword_at_rva if is_64 else pe.get_dword_at_rva cb_va = cb_rva - image_base if cb_rva > image_base else cb_rva count = 0 try: for i in range(100): ptr = read_ptr(cb_va + i * (8 if is_64 else 4)) if not ptr: break count += 1 except Exception: pass rpt.p(f"TLS callback count: {count}{' [FLAG: possible anti-debug via TLS callback]' if count else ''}") rpt.data["tls_callbacks"] = count def section_rich_header(pe, rpt: Report): rpt.h("Rich header") rich = getattr(pe, "RICH_HEADER", None) if rich is None: rpt.p("No Rich header.") rpt.data["rich_header"] = {"present": False} return values = getattr(rich, "values", []) or [] compiler_ids = len(values) // 2 rpt.p(f"Rich header present: yes ({compiler_ids} compiler-id entries)") rpt.data["rich_header"] = {"present": True, "compiler_id_count": compiler_ids} def analyze(path, as_json): rpt = Report() try: pe = pefile.PE(path, fast_load=True) except pefile.PEFormatError as e: print(f"Not a valid PE file: {e}", file=sys.stderr) return 1 except Exception as e: print(f"Failed to open file: {e}", file=sys.stderr) return 1 try: pe.parse_data_directories() except Exception as e: rpt.p(f"warning: parse_data_directories partially failed: {e}") with open(path, "rb") as f: overall_entropy = shannon_entropy(f.read()) is_dotnet = False try: is_dotnet = bool(section_file_info(pe, path, rpt)) except Exception as e: rpt.fail("file", e) sections = [] try: sections = section_sections(pe, rpt) except Exception as e: rpt.fail("sections", e) total_imports = 0 kernel32_loader_pattern = False try: total_imports, kernel32_loader_pattern = section_imports(pe, rpt) except Exception as e: rpt.fail("imports", e) try: section_packing_verdict(pe, rpt, overall_entropy, sections, total_imports, kernel32_loader_pattern, is_dotnet) except Exception as e: rpt.fail("packing_verdict", e) try: section_signature(pe, rpt) except Exception as e: rpt.fail("signature", e) try: section_resources(pe, rpt) except Exception as e: rpt.fail("resources", e) try: section_overlay(pe, rpt, path) except Exception as e: rpt.fail("overlay", e) try: section_pdb(pe, rpt) except Exception as e: rpt.fail("pdb", e) try: section_version_info(pe, rpt, path) except Exception as e: rpt.fail("version_info", e) try: section_tls(pe, rpt) except Exception as e: rpt.fail("tls", e) try: section_rich_header(pe, rpt) except Exception as e: rpt.fail("rich_header", e) if as_json: print(json.dumps(rpt.data, indent=2, default=str)) else: print("\n".join(rpt.lines)) return 0 def main(): parser = argparse.ArgumentParser(description="Static PE triage summary (never executes the sample).") parser.add_argument("path", help="Path to the PE file to analyze") parser.add_argument("--json", action="store_true", help="Output JSON instead of plain text") args = parser.parse_args() if not os.path.isfile(args.path): print(f"No such file: {args.path}", file=sys.stderr) sys.exit(1) sys.exit(analyze(args.path, args.json)) if __name__ == "__main__": main()