Adding unwind parsing and logging for pipeline integration

This commit is contained in:
toneillcodes
2026-06-18 16:50:40 -04:00
parent fbea61b4be
commit 38740889a6
3 changed files with 193 additions and 115 deletions
+102 -55
View File
@@ -1,95 +1,131 @@
import os
import csv
import argparse
import pefile
def dump_module_exports(dll_path, output_file=None, names_only=False):
def dump_module_exports(dll_path, log_file=None, console_output=True, names_only=False, process_name=None):
"""
Parses a local PE file, extracts the Export Address Table (EAT),
and displays a structured list of names, ordinals, and RVAs sorted by layout.
Explicitly flags and extracts Forwarded Exports to prevent stomping conflicts.
Parses a local PE file, extracts the Export Address Table (EAT),
calculates precise function sizes via .pdata exception entries,
and displays or logs a structured manifest sorted by layout.
"""
if not os.path.exists(dll_path):
print(f"[-] Error: Target file not found at '{dll_path}'")
return
dll_name = os.path.basename(dll_path)
target_process = process_name if process_name else "target_process.exe"
try:
# Load the PE file and explicitly force parsing of the export directory
# Load the PE file and explicitly force parsing of the directories
pe = pefile.PE(dll_path, fast_load=False)
if not hasattr(pe, 'DIRECTORY_ENTRY_EXPORT'):
print(f"[-] Error: '{os.path.basename(dll_path)}' has no Export Address Table (EAT).")
print(f"[-] Error: '{dll_name}' has no Export Address Table (EAT).")
return
exported_functions = []
# 1. Harvest exact runtime function sizes via SEH Exception Directory (.pdata)
pdata_sizes = {}
exception_dir_idx = pefile.DIRECTORY_ENTRY['IMAGE_DIRECTORY_ENTRY_EXCEPTION']
if len(pe.OPTIONAL_HEADER.DATA_DIRECTORY) > exception_dir_idx:
exception_dir = pe.OPTIONAL_HEADER.DATA_DIRECTORY[exception_dir_idx]
if exception_dir.VirtualAddress != 0 and exception_dir.Size != 0:
pe.parse_data_directories(directories=[exception_dir_idx])
if hasattr(pe, 'DIRECTORY_ENTRY_EXCEPTION'):
for entry in pe.DIRECTORY_ENTRY_EXCEPTION:
begin_rva = entry.struct.BeginAddress
end_rva = entry.struct.EndAddress
pdata_sizes[begin_rva] = end_rva - begin_rva
# Iterate over the parsed symbols in the export directory
# 2. Extract exports, filter forwarders, and calculate sizing configurations
exported_functions = []
for exp in pe.DIRECTORY_ENTRY_EXPORT.symbols:
# Derive name or fallback to ordinal string representation if un-exported by name
# Derive name or fallback to the new ordinal matching convention
if exp.name:
func_name = exp.name.decode('utf-8', errors='ignore')
else:
func_name = f"[No Symbol] Ordinal_{exp.ordinal}"
func_name = f"ordinal_{exp.ordinal}"
# CRITICAL HARDENING: Detect forwarded exports
# Detect forwarded exports
is_forwarded = False
forwarder_str = ""
if exp.forwarder:
is_forwarded = True
forwarder_str = exp.forwarder.decode('utf-8', errors='ignore')
rva = exp.address if exp.address else 0
# Look up size from unwind data; fallback to minimum block if missing or forwarded
calculated_size = pdata_sizes.get(rva, 64) if not is_forwarded else 0
exported_functions.append({
'name': func_name,
'ordinal': exp.ordinal,
'rva': exp.address if exp.address else 0,
'rva': rva,
'is_forwarded': is_forwarded,
'forwarder_target': forwarder_str
'forwarder_target': forwarder_str,
'size': calculated_size
})
# Sort functions sequentially by RVA so it maps linearly to memory layout
exported_functions_sorted = sorted(exported_functions, key=lambda x: x['rva'])
lines = []
if names_only:
# Generate a raw list of names only (filtering out forwarded assets or unnamed tokens)
for func in exported_functions_sorted:
if not func['name'].startswith("[No Symbol]") and not func['is_forwarded']:
lines.append(func['name'])
else:
# Generate full structured diagnostic manifest
lines.append("\n" + "="*115)
lines.append(f"📦 EXPORT ADDRESS TABLE MANIFEST: {os.path.basename(dll_path)}")
lines.append("="*115)
lines.append(f"{'Exported Symbol / Function Name':<45} | {'Ordinal':<8} | {'RVA / Status':<18} | {'Forwarding Target / Details'}")
lines.append("-"*115)
# --- CONDITION 1: DIRECT CONSOLE OUTPUT (DEFAULTS TO TRUE) ---
if console_output or names_only:
lines = []
if names_only:
# Generate a raw list of names only (filtering out anonymous tokens or forwarders)
for func in exported_functions_sorted:
if not func['name'].startswith("ordinal_") and not func['is_forwarded']:
lines.append(func['name'])
else:
# Generate full structured diagnostic manifest table
lines.append("\n" + "="*130)
lines.append(f"📦 EXPORT ADDRESS TABLE & CAPACITY MANIFEST: {dll_name}")
lines.append("="*130)
lines.append(f"{'Exported Symbol / Function Name':<45} | {'Ordinal':<8} | {'RVA Layout':<12} | {'Size':<10} | {'Forwarding Target / Details'}")
lines.append("-"*130)
for func in exported_functions_sorted:
if func['is_forwarded']:
rva_str = "[!] FORWARDED"
details = f"Points to -> {func['forwarder_target']}"
else:
rva_str = hex(func['rva']) if func['rva'] != 0 else "N/A"
details = "Native Code Body"
for func in exported_functions_sorted:
if func['is_forwarded']:
rva_str = "[!] FORWARDED"
size_str = "0 bytes"
details = f"Points to -> {func['forwarder_target']}"
else:
rva_str = hex(func['rva']) if func['rva'] != 0 else "N/A"
size_str = f"{func['size']} bytes"
details = "Native Code Body"
lines.append(f"{func['name']:<45} | {func['ordinal']:<8} | {rva_str:<18} | {details}")
lines.append(f"{func['name']:<45} | {func['ordinal']:<8} | {rva_str:<12} | {size_str:<10} | {details}")
lines.append("-"*115)
lines.append(f"[+] Successfully extracted {len(exported_functions_sorted)} exports from the EAT.")
lines.append("="*115 + "\n")
lines.append("-"*130)
lines.append(f"[+] Successfully extracted {len(exported_functions_sorted)} entries from the structural layout.")
lines.append("="*130 + "\n")
# Output to screen
for line in lines:
print(line)
for line in lines:
print(line)
# Write to file if specified
if output_file:
try:
with open(output_file, 'w', encoding='utf-8') as f:
for line in lines:
f.write(line + '\n')
print(f"[+] Successfully saved output to: '{output_file}'\n")
except Exception as file_err:
print(f"[-] Warning: Failed to write output to file: {file_err}")
# --- CONDITION 2: PIPELINE AUTOMATION CSV LOGGING ---
if log_file:
fieldnames = ['ModuleName', 'FullTextSectionSize', 'FunctionName', 'Offset', 'FuncSize']
with open(log_file, mode='w', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
for func in exported_functions_sorted:
# Skip forwarded assets in automated plans to prevent uncatchable stomp faults
if func['is_forwarded']:
continue
writer.writerow({
'ModuleName': dll_name,
'FullTextSectionSize': 0,
'FunctionName': func['name'],
'Offset': hex(func['rva']),
'FuncSize': func['size']
})
print(f"[+] Operational blueprint successfully committed to: '{log_file}'")
except pefile.PEFormatError:
print(f"[-] Error: '{dll_path}' is not a valid Portable Executable file.")
@@ -97,10 +133,21 @@ def dump_module_exports(dll_path, output_file=None, names_only=False):
print(f"[-] An unexpected processing error occurred: {e}")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Extract and map the complete Export Address Table layout from a target PE module.")
parser.add_argument("-f", "--file", type=str, required=True, help="Path to the target DLL on disk.")
parser.add_argument("-o", "--output", type=str, default=None, help="Optional path to a text file to write the output to.")
parser.add_argument("--names-only", action="store_true", help="Output a clean list of function names only, omitting headers, RVAs, and ordinals.")
parser = argparse.ArgumentParser(description="Extract and map export table capacities with structural unwind parsing.")
parser.add_argument("-f", "--file", type=str, required=True, help="Path to the target DLL on disk.")
# Execution Flow Control Switches
parser.add_argument("-l", "--log", nargs='?', const="stability_blueprint.csv", default=None,
help="Enable pipeline CSV generation. Optionally specify a file path destination.")
parser.add_argument("--names-only", action="store_true", help="Output a clean list of string function names only.")
# Control console formatting behavior (Defaulting to True, switchable via --no-console)
parser.add_argument("-c", "--console", dest="console", action="store_true", default=True, help="Print structured table matrix straight to terminal (Default).")
parser.add_argument("--no-console", dest="console", action="store_false", help="Disable text table generation entirely for background runs.")
args = parser.parse_args()
dump_module_exports(args.file, args.output, args.names_only)
dump_module_exports(
dll_path=args.file,
log_file=args.log,
console_output=args.console,
names_only=args.names_only
)
+59 -60
View File
@@ -1,4 +1,5 @@
import os
import csv
import argparse
import pefile
@@ -33,26 +34,27 @@ def parse_flat_file(file_path, list_type_label):
return names_set
def find_phantom_dll_candidates(target_dir, min_file_size_mb, min_text_section_size, excluded_dlls, included_dlls):
def find_phantom_dll_candidates(target_dir, min_file_size_mb, min_text_section_size, excluded_dlls, included_dlls, log_file=None, console_output=True):
"""
Scans a directory for DLLs matching structural criteria.
Supports exclusion filters (blacklists) and targeted scope filters (whitelists).
Supports exclusion filters, targeted scope filters, and conditional pipeline logging.
"""
min_file_size_bytes = min_file_size_mb * 1024 * 1024
candidates = []
print(f"[*] Scanning Target Directory: '{target_dir}'")
print(f"[*] Filtering for Files : > {min_file_size_mb}MB")
print(f"[*] Required .text Space : {hex(min_text_section_size)} bytes")
if included_dlls:
print(f"[*] Targeted Include Filter : Active ({len(included_dlls)} specific targets allowed)")
if excluded_dlls:
print(f"[*] Exclusion Filter : Active ({len(excluded_dlls)} modules blacklisted)")
if console_output:
print(f"[*] Scanning Target Directory: '{target_dir}'")
print(f"[*] Filtering for Files : > {min_file_size_mb}MB")
print(f"[*] Required .text Space : {hex(min_text_section_size)} bytes")
print("-" * 140)
print(f"{'Full File Path':<85} | {'File Size (MB)':<15} | {'Size of .text':<15} | {'Virtual Address':<15}")
print("-" * 140)
if included_dlls:
print(f"[*] Targeted Include Filter : Active ({len(included_dlls)} specific targets allowed)")
if excluded_dlls:
print(f"[*] Exclusion Filter : Active ({len(excluded_dlls)} modules blacklisted)")
print("-" * 140)
print(f"{'Full File Path':<85} | {'File Size (MB)':<15} | {'Size of .text':<15} | {'Virtual Address':<15}")
print("-" * 140)
for root, _, files in os.walk(target_dir):
for file in files:
@@ -60,11 +62,11 @@ def find_phantom_dll_candidates(target_dir, min_file_size_mb, min_text_section_s
if not file_lower.endswith('.dll'):
continue
# Gate 1: If an include list is provided, ignore everything else
# Gate 1: Include lookup list filter bounds
if included_dlls and (file_lower not in included_dlls):
continue
# Gate 2: If an exclude list is provided, drop any matching items
# Gate 2: Exclude lookup blacklist filter bounds
if excluded_dlls and (file_lower in excluded_dlls):
continue
@@ -85,62 +87,57 @@ def find_phantom_dll_candidates(target_dir, min_file_size_mb, min_text_section_s
if v_size >= min_text_section_size:
file_size_mb = file_size / (1024 * 1024)
print(f"{file_path:<85} | {file_size_mb:<15.2f} | {hex(v_size):<15} | {hex(section.VirtualAddress):<15}")
if console_output:
print(f"{file_path:<85} | {file_size_mb:<15.2f} | {hex(v_size):<15} | {hex(section.VirtualAddress):<15}")
candidates.append({
'path': file_path,
'file_size': file_size,
'text_virtual_size': v_size
'Name': file, # Base filename matches list-process-dlls paradigm expect layouts
'TextSectionSize': v_size
})
break
except (pefile.PEFormatError, PermissionError, FileNotFoundError):
continue
print("-" * 140)
print(f"[*] Found {len(candidates)} potential candidates matching the criteria.")
if console_output:
print("-" * 140)
print(f"[*] Found {len(candidates)} potential candidates matching the criteria.")
# --- PIPELINE SEAMLESS AUTOMATION EXPORT ---
if log_file and candidates:
# Field structure exactly matching list-process-dlls layout configuration needs
fieldnames = ['Name', 'TextSectionSize']
try:
with open(log_file, mode='w', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
for cand in candidates:
writer.writerow({
'Name': cand['Name'],
'TextSectionSize': cand['TextSectionSize']
})
print(f"[+] Operational structural blueprint saved to: '{log_file}'")
except Exception as file_err:
print(f"[-] Error: Failed to write pipeline target spreadsheet: {file_err}")
return candidates
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Find DLL candidates with targeted tracking and exclusion parameters.")
# Space required
parser.add_argument(
"size",
type=auto_int,
help="The total required size of the .text section (e.g., 0x80000)"
)
# Positionals & core targets
parser.add_argument("size", type=auto_int, help="The total required size of the .text section (e.g., 0x80000)")
parser.add_argument("-d", "--dir", type=str, default=r"C:\Windows\System32", help="Target directory to scan (default: C:\\Windows\\System32)")
parser.add_argument("-m", "--min-size", type=float, default=1.0, help="Minimum file size on disk in MB (default: 1.0)")
# Directory to scan
parser.add_argument(
"-d", "--dir",
type=str,
default=r"C:\Windows\System32",
help="Target directory to scan (default: C:\\Windows\\System32)"
)
# Minimum size of image
parser.add_argument(
"-m", "--min-size",
type=float,
default=1.0,
help="Minimum file size on disk in MB (default: 1.0)"
)
# Exclude list
parser.add_argument(
"-x", "--exclude",
type=str,
default=None,
help="Path to a text file containing specific DLLs to EXCLUDE"
)
# Flat configurations files switches
parser.add_argument("-x", "--exclude", type=str, default=None, help="Path to a text file containing specific DLLs to EXCLUDE")
parser.add_argument("-i", "--include", type=str, default=None, help="Path to a text file containing specific DLLs to INCLUDE")
# Include list
parser.add_argument(
"-i", "--include",
type=str,
default=None,
help="Path to a text file containing specific DLLs to INCLUDE"
)
# Pipeline output and terminal display switches
parser.add_argument("-l", "--log", type=str, default=None, help="Destination pipeline tracking file path to export CSV output parameters.")
parser.add_argument("-c", "--console", dest="console", action="store_true", default=True, help="Print structured layout results table directly to screen terminal (Default).")
parser.add_argument("--no-console", dest="console", action="store_false", help="Suppress terminal visual representation during backend background tasks loops.")
args = parser.parse_args()
@@ -148,7 +145,7 @@ if __name__ == "__main__":
print(f"[-] Error: '{args.dir}' is not a valid directory.")
exit(1)
# Parse both filters independently
# Parse flat filters definitions
excluded_modules = parse_flat_file(args.exclude, "EXCLUDE_MODULES")
included_modules = parse_flat_file(args.include, "INCLUDE_MODULES")
@@ -157,5 +154,7 @@ if __name__ == "__main__":
min_file_size_mb=args.min_size,
min_text_section_size=args.size,
excluded_dlls=excluded_modules,
included_dlls=included_modules
)
included_dlls=included_modules,
log_file=args.log,
console_output=args.console
)
+32
View File
@@ -0,0 +1,32 @@
#!/usr/bin/python
import sys
# Check if both input and output filenames were provided as arguments
if len(sys.argv) < 3:
print(f"Usage: {sys.argv[0]} <input_file> <output_file>")
sys.exit(1)
input_filename = sys.argv[1]
output_filename = sys.argv[2]
# Read the entire large payload into memory
with open(input_filename, "rb") as input_file:
payload = input_file.read()
# Convert the binary data into a C-compatible integer array format
# This completely avoids C1091 by eliminating string literals entirely ("")
lines = []
for i in range(0, len(payload), 12): # 12 bytes per line keeps it clean
chunk = payload[i:i+12]
hex_line = ", ".join(f"0x{byte:02x}" for byte in chunk)
lines.append(f" {hex_line}")
# Join all lines together with commas and newlines
array_content = ",\n".join(lines)
# Write the final array to the output file
with open(output_filename, "w") as output_file:
output_file.write(f"// Payload size: {len(payload)} bytes\n")
output_file.write("unsigned char buf[] = {\n")
output_file.write(array_content)
output_file.write("\n};\n")