USER
User: Review the python code, what does the code do?
```
# main.py
import os
from config import BINARY_PATH, CPP_OUTPUT_DIR
from utils.project_info import gather_project_metadata
from utils.decompilation import decompile_functions
from utils.validation import validate_function
from utils.verification import verify_function
from utils.db_utils import save_to_database, close_database
from utils.generate_file_structure import generate_files_from_db
def main():
gather_project_metadata(BINARY_PATH)
os.makedirs(CPP_OUTPUT_DIR, exist_ok=True)
decompiled_functions = decompile_functions(BINARY_PATH)
for func_name, func_data in decompiled_functions.items():
if validate_function(func_data):
save_to_database(func_name, func_data['offset'], func_data['assembly_code'], func_data['decompiled_code'], "")
verified_functions = verify_function(BINARY_PATH, decompiled_functions)
feedback_loop(decompiled_functions, verified_functions)
generate_files_from_db()
close_database()
if __name__ == "__main__":
main()
# config.py
# Base URL for the Ollama API running on WSL2
OLLAMA_BASE_URL = "http://172.31.72.252:11434"
# Model to use in Ollama
LLM_MODEL = "ollama/qwen2.5:14b"
# Debug mode toggle
DEBUG_MODE = True
# Paths for project files and directories
BINARY_PATH = r"Civ5XP" # Path to the binary file for decompilation
CPP_OUTPUT_DIR = "output" # Directory for C++ output files
LOG_FILE = "logs/debug.log" # Log file path
DATABASE_PATH = "binary_decompilation.db" # Database file path
# Decompilation and C++ generation settings
MAX_CONTEXT_LENGTH = 128000 # Max context length for LLM input
MAX_TOKENS = 8192 # Max tokens for LLM output
CXX_STANDARD = 17 # C++ standard for generated code
# Function call model configuration for LLM
MODEL_FUNCTION_CALL_SETTINGS = {
"name": "generate_cpp",
"parameters": [
{"name": "cpp_code", "type": "string"},
{"name": "warnings", "type": "list"},
{"name": "feedback", "type": "string"}
]
}
# CMake file generation options
CMAKE_MINIMUM_VERSION = "3.10"
TARGET_NAME = "DecompiledProject" # Target name in CMakeLists.txt
ADDITIONAL_LIBRARIES = ["libA", "libB"] # Additional libraries to link in CMakeLists.txt
LLVM_PATH = "/usr/local/llvm" # Adjust to your actual LLVM path if needed
INCLUDE_DIRECTORIES = ["include", "/path/to/other/includes"]
```
```
# utils/decompilation.py
from utils.verification import verify_function, compile_cpp_to_llvm_ir
from utils.validation import validate_function
from utils.feedback_loop import feedback_loop
def decompile_functions(binary_file=BINARY_PATH):
project_metadata = gather_project_metadata(binary_file)
function_summaries = {}
with pyhidra.open_program(binary_file) as flat_api:
program = flat_api.getCurrentProgram()
listing = program.getListing()
decompiler = DecompInterface()
decompiler.openProgram(program)
for function in listing.getFunctions(True):
function_name = function.getName()
function_offset = function.getEntryPoint().getOffset()
db_entry = load_function_from_database(function_name)
if db_entry:
logging.debug(f"Function '{function_name}' already in database.")
continue
try:
results = decompiler.decompileFunction(function, 0, TaskMonitor.DUMMY)
if results and results.decompiledFunction is not None:
decompiled_code = results.getDecompiledFunction().getC()
assembly_code = str(function.getBody())
cpp_code = generate_cpp_code(decompiled_code, function_name)
# Decompile function and collect metadata
function_metadata = {
"name": function_name,
"offset": function_offset,
"assembly_code": assembly_code,
"decompiled_code": decompiled_code,
"symbol_names": [symbol.getName() for symbol in function.getSymbols()],
"comments": extract_comments_from_function(function),
"cpp_code": cpp_code
}
# Store to database after validation
if validate_function(function_metadata):
save_to_database(function_name, function_metadata, 'decompilation_output')
function_summaries[function_name] = function_metadata
# Verification step - LLVM IR and Control Flow Graph matching
llvm_ir = compile_cpp_to_llvm_ir(cpp_code)
verification_result = verify_function(binary_file, {function_name: function_metadata})
if not verification_result:
logging.warning(f"Verification failed for '{function_name}'. Triggering feedback loop.")
feedback_loop({function_name: function_metadata}, function_summaries)
else:
logging.warning(f"Validation failed for '{function_name}'.")
else:
logging.error(f"Failed decompiling '{function_name}'.")
except Exception as e:
logging.error(f"Error processing '{function_name}': {str(e)}")
close_database()
return function_summaries
```
```
# utils/verification.py
import angr
import llvmlite.binding as llvm
import logging
import subprocess
import tempfile
from config import DEBUG_MODE
from utils.db_utils import load_project_metadata
logging.basicConfig(filename="logs/debug.log", level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def verify_function(binary_path, functions_data):
project_metadata = load_project_metadata()
min_address = int(project_metadata.get("min_address", "0"), 16)
max_address = int(project_metadata.get("max_address", "FFFFFFFF"), 16)
for func_name, func_data in functions_data.items():
cpp_code = func_data.get("cpp_code")
if cpp_code:
llvm_ir = compile_cpp_to_llvm_ir(cpp_code)
if llvm_ir:
result = compare_with_binary(binary_path, func_name, llvm_ir, min_address, max_address)
if result:
logging.info(f"Verification passed for '{func_name}'.")
else:
logging.warning(f"Verification failed for '{func_name}'.")
def compile_cpp_to_llvm_ir(cpp_code):
try:
with tempfile.NamedTemporaryFile(suffix=".cpp", delete=False) as cpp_file:
cpp_file.write(cpp_code.encode())
cpp_filename = cpp_file.name
llvm_filename = cpp_filename.replace(".cpp", ".ll")
subprocess.run(["clang", "-emit-llvm", "-S", cpp_filename, "-o", llvm_filename], check=True)
with open(llvm_filename, "r") as llvm_ir_file:
llvm_ir = llvm_ir_file.read()
return llvm_ir
except subprocess.CalledProcessError as e:
logging.error(f"Clang compilation failed: {str(e)}")
return None
except Exception as e:
logging.error(f"Error compiling C++ to LLVM IR: {str(e)}")
return None
def compare_with_binary(binary_path, func_name, llvm_ir, min_address, max_address):
try:
project = angr.Project(binary_path, auto_load_libs=False)
cfg = project.analyses.CFGFast()
llvm_func_addresses = extract_function_addresses_from_llvm(llvm_ir, func_name)
for address in llvm_func_addresses:
if not (min_address <= address <= max_address):
logging.warning(f"Address {address} for '{func_name}' out of valid range.")
continue
binary_func = cfg.kb.functions.get_by_addr(address)
if binary_func is None:
logging.warning(f"Function '{func_name}' at address {address} not found in binary.")
return False
return True
except Exception as e:
logging.error(f"Error comparing LLVM IR with binary: {str(e)}")
return False
def extract_function_addresses_from_llvm(llvm_ir, func_name):
addresses = []
for line in llvm_ir.splitlines():
if func_name in line and "define" in line:
# Use a more accurate regex or parser for proper extraction
# Assuming format like `@func_name = external addrspace(0) constant i32 0x...`
address_match = re.search(r'0x[0-9A-Fa-f]+', line)
if address_match:
address = int(address_match.group(0), 16)
addresses.append(address)
return addresses
```
```
# utils/validation.py
import jsonschema
import logging
import json
from config import DEBUG_MODE
logging.basicConfig(filename="logs/debug.log", level=logging.DEBUG if DEBUG_MODE else logging.INFO)
with open("schemas/function_schema.json") as f:
function_schema = json.load(f)
def validate_function(function_data):
try:
jsonschema.validate(instance=function_data, schema=function_schema)
logging.debug(f"Validation passed: {function_data.get('name', 'unknown')}")
return True
except jsonschema.ValidationError as e:
logging.error(f"Validation error: {e}")
return False
```
```
# utils/db_utils.py
import sqlite3
from pathlib import Path
db_path = Path("binary_decompilation.db")
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Functions table
cursor.execute('''
CREATE TABLE IF NOT EXISTS functions (
function_id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE,
address TEXT,
entry_point TEXT,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Decompilation Output table
cursor.execute('''
CREATE TABLE IF NOT EXISTS decompilation_output (
function_id INTEGER,
assembly_code TEXT,
pseudo_c_code TEXT,
last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (function_id) REFERENCES functions(function_id)
)
''')
# C++ Generation table
cursor.execute('''
CREATE TABLE IF NOT EXISTS cpp_generation (
function_id INTEGER,
cpp_code TEXT,
warnings TEXT,
feedback TEXT,
validation_status TEXT,
last_generated TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (function_id) REFERENCES functions(function_id)
)
''')
# Verification and Feedback table
cursor.execute('''
CREATE TABLE IF NOT EXISTS verification_feedback (
function_id INTEGER,
discrepancies TEXT,
refined_prompt TEXT,
verification_status TEXT,
last_verified TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (function_id) REFERENCES functions(function_id)
)
''')
conn.commit()
# Project Metadata table
cursor.execute('''
CREATE TABLE IF NOT EXISTS project_metadata (
project_id INTEGER PRIMARY KEY AUTOINCREMENT,
project_file_name TEXT,
last_modified TEXT,
readonly BOOLEAN,
program_name TEXT,
language_id TEXT,
compiler_id TEXT,
processor TEXT,
endian TEXT,
address_size INTEGER,
min_address TEXT,
max_address TEXT,
num_bytes INTEGER,
num_memory_blocks INTEGER,
num_instructions INTEGER,
num_defined_data INTEGER,
num_functions INTEGER,
num_symbols INTEGER,
num_data_types INTEGER,
analyzed BOOLEAN,
created_with_ghidra_version TEXT,
file_type TEXT,
file_location TEXT,
elf_original_image_base TEXT,
relocatable BOOLEAN,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Missing Libraries table
cursor.execute('''
CREATE TABLE IF NOT EXISTS missing_libraries (
library_id INTEGER PRIMARY KEY AUTOINCREMENT,
project_id INTEGER,
library_name TEXT,
last_checked TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (project_id) REFERENCES project_metadata(project_id)
)
''')
conn.commit()
def save_project_metadata(metadata):
fields = ', '.join(metadata.keys())
placeholders = ', '.join(['?'] * len(metadata))
values = tuple(metadata.values())
cursor.execute(f'''
INSERT INTO project_metadata ({fields})
VALUES ({placeholders})
''', values)
conn.commit()
def save_missing_library(project_id, library_name):
cursor.execute('''
INSERT INTO missing_libraries (project_id, library_name)
VALUES (?, ?)
''', (project_id, library_name))
conn.commit()
def save_to_database(function_id, data, table):
try:
fields = ', '.join(data.keys())
placeholders = ', '.join(['?'] * len(data))
values = tuple(data.values())
cursor.execute(f'''
INSERT OR REPLACE INTO {table} ({fields})
VALUES ({placeholders})
''', values)
conn.commit()
except sqlite3.Error as e:
print(f"Database error: {e}")
def load_function_from_database(function_name):
cursor.execute('SELECT * FROM functions WHERE name = ?', (function_name,))
return cursor.fetchone()
def close_database():
conn.close()
# context manager for database operations
from contextlib import contextmanager
@contextmanager
def get_db_cursor():
conn = sqlite3.connect(db_path)
try:
yield conn.cursor()
finally:
conn.commit()
conn.close()
```
```
# utils/feedback_loop.py
import logging
from utils.db_utils import save_to_database
from config import DEBUG_MODE
logging.basicConfig(filename="logs/debug.log", level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def log_discrepancies(function_id, discrepancy_details):
logging.debug(f"Logging discrepancy for function_id '{function_id}': {discrepancy_details}")
save_to_database(function_id, {"discrepancies": str(discrepancy_details)}, "verification_feedback")
def refine_prompts(discrepancies):
if not discrepancies:
logging.debug("No discrepancies found; no prompt refinement necessary.")
return
for func_id, details in discrepancies:
logging.debug(f"Refining prompts based on discrepancy for function_id '{func_id}'")
# Adjustments to prompt structure or content based on discrepancy type
if "type_mismatch" in details:
prompt_modifier = "Ensure strict type compliance in generated C++."
elif "missing_code" in details:
prompt_modifier = "Double-check all control flows and branches."
refined_prompt = {
"role": "user",
"content": f"Refine the C++ output based on these observations:\n{details}\n{prompt_modifier}"
}
save_to_database(func_id, {"refined_prompt": str(refined_prompt)}, "verification_feedback")
logging.debug(f"Updated prompt for function_id '{func_id}': {refined_prompt}")
def feedback_loop(decompiled_functions, verified_functions):
discrepancies = []
for func_id, details in decompiled_functions.items():
if func_id not in verified_functions or verified_functions[func_id]["cpp_code"] != details["cpp_code"]:
discrepancy_details = {"decompiled": details, "verified": verified_functions.get(func_id)}
discrepancies.append((func_id, discrepancy_details))
log_discrepancies(func_id, discrepancy_details)
refine_prompts(discrepancies)
```
```
# utils/project_info.py
import os
import logging
import pyhidra
from utils.db_utils import save_project_metadata, save_missing_library
logging.basicConfig(filename="logs/project_info.log", level=logging.INFO)
def gather_project_metadata(binary_path):
with pyhidra.open_program(binary_path) as flat_api:
program = flat_api.getCurrentProgram()
metadata = {
"project_file_name": program.getDomainFile().getName(),
"last_modified": program.getModificationDate().toString(),
"readonly": program.isReadonly(),
"program_name": program.getName(),
"language_id": program.getLanguageID().toString(),
"compiler_id": program.getCompilerSpec().getCompilerSpecID().toString(),
"processor": program.getLanguage().getProcessor().toString(),
"endian": program.getLanguage().isBigEndian(),
"address_size": program.getDefaultPointerSize(),
"min_address": program.getMinAddress().toString(),
"max_address": program.getMaxAddress().toString(),
"num_bytes": program.getMemory().getNumAddresses(),
"num_memory_blocks": program.getMemoryBlockCount(),
"num_instructions": program.getListing().getNumInstructions(),
"num_defined_data": program.getListing().getNumDefinedData(),
"num_functions": program.getFunctionManager().getFunctionCount(),
"num_symbols": program.getSymbolTable().getNumSymbols(),
"num_data_types": len(program.getDataTypeManager().getAllDataTypes()),
"analyzed": program.isAnalyzed(),
"created_with_ghidra_version": program.getVersion(),
"file_type": program.getExecutableFormat(),
"file_location": program.getExecutablePath(),
"elf_original_image_base": program.getExecutableBase(),
"relocatable": program.isRelocatable()
}
save_project_metadata(metadata)
# Log and save required libraries
required_libs = program.getMemory().getExternalLibraries()
for lib in required_libs:
if not os.path.exists(lib):
logging.warning(f"Missing library: {lib}")
save_missing_library(lib)
logging.info("Project metadata and library information gathered and saved.")
```
```
# utils/cpp_generator.py
import logging
from config import LLM_MODEL, DEBUG_MODE, MAX_CONTEXT_LENGTH, MAX_TOKENS, MODEL_FUNCTION_CALL_SETTINGS, LOG_FILE
from litellm import completion
logging.basicConfig(filename=LOG_FILE, level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def generate_cpp_code(pseudo_c_code, function_name, symbol_names="", comments=""):
try:
# Include symbols and comments in the prompt for context-awareness
prompt_content = f"Convert the following pseudo-C code to structured C++:\n{pseudo_c_code}"
if symbol_names or comments:
prompt_content += f"\n\nSymbols:\n{symbol_names}\nComments:\n{comments}"
cpp_code_response = completion(
model=LLM_MODEL,
messages=[{
"role": "user",
"content": prompt_content
}],
format="json",
max_context_length=MAX_CONTEXT_LENGTH,
max_tokens=MAX_TOKENS,
function_call=MODEL_FUNCTION_CALL_SETTINGS
)
# Similarity check and feedback trigger
cpp_code = cpp_code_response.get('output', {}).get('cpp_code', '')
if not check_similarity(pseudo_c_code, cpp_code):
logging.warning(f"Similarity check failed for '{function_name}'. Triggering feedback loop.")
feedback_loop({function_name: function_metadata})
return cpp_code
except Exception as e:
logging.error(f"Error generating C++ code for function '{function_name}': {str(e)}")
return None
```
```
# utils/generate_file_structure.py
import os
import logging
from config import (DATABASE_PATH, CPP_OUTPUT_DIR, CMAKE_MINIMUM_VERSION, TARGET_NAME, CXX_STANDARD,
ADDITIONAL_LIBRARIES, LLVM_PATH, INCLUDE_DIRECTORIES, DEBUG_MODE, LOG_FILE)
logging.basicConfig(filename=LOG_FILE, level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def generate_files_from_db():
# Connect to database and fetch functions
with sqlite3.connect(DATABASE_PATH) as conn:
cursor = conn.cursor()
project_metadata = gather_project_metadata()
cursor.execute("SELECT function_id, name, cpp_code, offset, signature FROM cpp_generation WHERE validation_status = 'verified'")
functions = cursor.fetchall()
header_content = {}
source_content = {}
for func_id, func_name, cpp_code, offset, signature in functions:
header_file, source_file = determine_file_structure(func_name, project_metadata)
if header_file not in header_content:
header_content[header_file] = ""
header_content[header_file] += f"{signature};\n"
if source_file not in source_content:
source_content[source_file] = ""
source_content[source_file] += cpp_code + "\n"
# Write header and source files
for filename, content in header_content.items():
with open(os.path.join(CPP_OUTPUT_DIR, "include", filename), "w") as f:
f.write("#pragma once\n\n" + content)
for filename, content in source_content.items():
with open(os.path.join(CPP_OUTPUT_DIR, "src", filename), "w") as f:
f.write(content)
generate_cmake_file(project_metadata)
def determine_file_structure(func_name, project_metadata):
module = project_metadata.get("modules", {}).get(func_name)
namespace = project_metadata.get("namespaces", {}).get(func_name)
if module:
module_dir = os.path.join(CPP_OUTPUT_DIR, "src", module)
os.makedirs(module_dir, exist_ok=True)
header_file = os.path.join(module, f"{func_name}.h")
source_file = os.path.join(module, f"{func_name}.cpp")
elif namespace:
namespace_dir = os.path.join(CPP_OUTPUT_DIR, "src", namespace)
os.makedirs(namespace_dir, exist_ok=True)
header_file = os.path.join(namespace, f"{func_name}.h")
source_file = os.path.join(namespace, f"{func_name}.cpp")
else:
header_file = f"{func_name}.h"
source_file = f"{func_name}.cpp"
return header_file, source_file
def generate_cmake_file(project_metadata):
# Build CMake content
library_includes = "\n".join([f"target_link_libraries({TARGET_NAME} {lib})" for lib in ADDITIONAL_LIBRARIES])
include_directories = "\n".join([f"include_directories({dir})" for dir in INCLUDE_DIRECTORIES])
cmake_content = f"""
cmake_minimum_required(VERSION {CMAKE_MINIMUM_VERSION})
project({TARGET_NAME})
set(CMAKE_CXX_STANDARD {CXX_STANDARD})
{include_directories}
# Add sources
file(GLOB SOURCES "src/**/*.cpp")
# Define executable
add_executable({TARGET_NAME} ${{SOURCES}})
# Link libraries
{library_includes}
# LLVM configuration (if needed)
if (EXISTS "{LLVM_PATH}")
find_package(LLVM REQUIRED PATHS "{LLVM_PATH}")
target_include_directories({TARGET_NAME} PRIVATE ${{LLVM_INCLUDE_DIRS}})
target_link_libraries({TARGET_NAME} ${{LLVM_LIBS}})
add_definitions(${{LLVM_DEFINITIONS}})
endif()
"""
with open(os.path.join(CPP_OUTPUT_DIR, "CMakeLists.txt"), "w") as cmake_file:
cmake_file.write(cmake_content)
```
```
function_schema.json:
{
"type": "object",
"properties": {
"name": {"type": "string"},
"address": {"type": "string"},
"offset": {"type": "integer"},
"instructions": {
"type": "array",
"items": {"type": "string"}
},
"pseudo_code": {"type": "string"},
"cpp_code": {"type": "string"}
},
"required": ["name", "address", "instructions", "pseudo_code", "cpp_code"]
}
```
```plaintext
# main.py
import os
from config import BINARY_PATH, CPP_OUTPUT_DIR
from utils.project_info import gather_project_metadata
from utils.decompilation import decompile_functions
from utils.validation import validate_function
from utils.verification import verify_function
from utils.db_utils import save_to_database, close_database
from utils.generate_file_structure import generate_files_from_db
def main():
gather_project_metadata(BINARY_PATH)
os.makedirs(CPP_OUTPUT_DIR, exist_ok=True)
decompiled_functions = decompile_functions(BINARY_PATH)
for func_name, func_data in decompiled_functions.items():
if validate_function(func_data):
save_to_database(func_name, func_data['offset'], func_data['assembly_code'], func_data['decompiled_code'], "")
verified_functions = verify_function(BINARY_PATH, decompiled_functions)
feedback_loop(decompiled_functions, verified_functions)
generate_files_from_db()
close_database()
if __name__ == "__main__":
main()
# config.py
# Base URL for the Ollama API running on WSL2
OLLAMA_BASE_URL = "http://172.31.72.252:11434"
# Model to use in Ollama
LLM_MODEL = "ollama/qwen2.5:14b"
# Debug mode toggle
DEBUG_MODE = True
# Paths for project files and directories
BINARY_PATH = r"Civ5XP" # Path to the binary file for decompilation
CPP_OUTPUT_DIR = "output" # Directory for C++ output files
LOG_FILE = "logs/debug.log" # Log file path
DATABASE_PATH = "binary_decompilation.db" # Database file path
# Decompilation and C++ generation settings
MAX_CONTEXT_LENGTH = 128000 # Max context length for LLM input
MAX_TOKENS = 8192 # Max tokens for LLM output
CXX_STANDARD = 17 # C++ standard for generated code
# Function call model configuration for LLM
MODEL_FUNCTION_CALL_SETTINGS = {
"name": "generate_cpp",
"parameters": [
{"name": "cpp_code", "type": "string"},
{"name": "warnings", "type": "list"},
{"name": "feedback", "type": "string"}
]
}
# CMake file generation options
CMAKE_MINIMUM_VERSION = "3.10"
TARGET_NAME = "DecompiledProject" # Target name in CMakeLists.txt
ADDITIONAL_LIBRARIES = ["libA", "libB"] # Additional libraries to link in CMakeLists.txt
LLVM_PATH = "/usr/local/llvm" # Adjust to your actual LLVM path if needed
INCLUDE_DIRECTORIES = ["include", "/path/to/other/includes"]
# utils/decompilation.py
from utils.verification import verify_function, compile_cpp_to_llvm_ir
from utils.validation import validate_function
from utils.feedback_loop import feedback_loop
def decompile_functions(binary_file=BINARY_PATH):
project_metadata = gather_project_metadata(binary_file)
function_summaries = {}
with pyhidra.open_program(binary_file) as flat_api:
program = flat_api.getCurrentProgram()
listing = program.getListing()
decompiler = DecompInterface()
decompiler.openProgram(program)
for function in listing.getFunctions(True):
function_name = function.getName()
function_offset = function.getEntryPoint().getOffset()
db_entry = load_function_from_database(function_name)
if db_entry:
logging.debug(f"Function '{function_name}' already in database.")
continue
try:
results = decompiler.decompileFunction(function, 0, TaskMonitor.DUMMY)
if results and results.decompiledFunction is not None:
decompiled_code = results.getDecompiledFunction().getC()
assembly_code = str(function.getBody())
cpp_code = generate_cpp_code(decompiled_code, function_name)
# Decompile function and collect metadata
function_metadata = {
"name": function_name,
"offset": function_offset,
"assembly_code": assembly_code,
"decompiled_code": decompiled_code,
"symbol_names": [symbol.getName() for symbol in function.getSymbols()],
"comments": extract_comments_from_function(function),
"cpp_code": cpp_code
}
# Store to database after validation
if validate_function(function_metadata):
save_to_database(function_name, function_metadata, 'decompilation_output')
function_summaries[function_name] = function_metadata
# Verification step - LLVM IR and Control Flow Graph matching
llvm_ir = compile_cpp_to_llvm_ir(cpp_code)
verification_result = verify_function(binary_file, {function_name: function_metadata})
if not verification_result:
logging.warning(f"Verification failed for '{function_name}'. Triggering feedback loop.")
feedback_loop({function_name: function_metadata}, function_summaries)
else:
logging.warning(f"Validation failed for '{function_name}'.")
else:
logging.error(f"Failed decompiling '{function_name}'.")
except Exception as e:
logging.error(f"Error processing '{function_name}': {str(e)}")
close_database()
return function_summaries
# utils/verification.py
import angr
import llvmlite.binding as llvm
import logging
import subprocess
import tempfile
from config import DEBUG_MODE
from utils.db_utils import load_project_metadata
logging.basicConfig(filename="logs/debug.log", level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def verify_function(binary_path, functions_data):
project_metadata = load_project_metadata()
min_address = int(project_metadata.get("min_address", "0"), 16)
max_address = int(project_metadata.get("max_address", "FFFFFFFF"), 16)
for func_name, func_data in functions_data.items():
cpp_code = func_data.get("cpp_code")
if cpp_code:
llvm_ir = compile_cpp_to_llvm_ir(cpp_code)
if llvm_ir:
result = compare_with_binary(binary_path, func_name, llvm_ir, min_address, max_address)
if result:
logging.info(f"Verification passed for '{func_name}'.")
else:
logging.warning(f"Verification failed for '{func_name}'.")
def compile_cpp_to_llvm_ir(cpp_code):
try:
with tempfile.NamedTemporaryFile(suffix=".cpp", delete=False) as cpp_file:
cpp_file.write(cpp_code.encode())
cpp_filename = cpp_file.name
llvm_filename = cpp_filename.replace(".cpp", ".ll")
subprocess.run(["clang", "-emit-llvm", "-S", cpp_filename, "-o", llvm_filename], check=True)
with open(llvm_filename, "r") as llvm_ir_file:
llvm_ir = llvm_ir_file.read()
return llvm_ir
except subprocess.CalledProcessError as e:
logging.error(f"Clang compilation failed: {str(e)}")
return None
except Exception as e:
logging.error(f"Error compiling C++ to LLVM IR: {str(e)}")
return None
def compare_with_binary(binary_path, func_name, llvm_ir, min_address, max_address):
try:
project = angr.Project(binary_path, auto_load_libs=False)
cfg = project.analyses.CFGFast()
llvm_func_addresses = extract_function_addresses_from_llvm(llvm_ir, func_name)
for address in llvm_func_addresses:
if not (min_address <= address <= max_address):
logging.warning(f"Address {address} for '{func_name}' out of valid range.")
continue
binary_func = cfg.kb.functions.get_by_addr(address)
if binary_func is None:
logging.warning(f"Function '{func_name}' at address {address} not found in binary.")
return False
return True
except Exception as e:
logging.error(f"Error comparing LLVM IR with binary: {str(e)}")
return False
def extract_function_addresses_from_llvm(llvm_ir, func_name):
addresses = []
for line in llvm_ir.splitlines():
if func_name in line and "define" in line:
# Use a more accurate regex or parser for proper extraction
# Assuming format like `@func_name = external addrspace(0) constant i32 0x...`
address_match = re.search(r'0x[0-9A-Fa-f]+', line)
if address_match:
address = int(address_match.group(0), 16)
addresses.append(address)
return addresses
# utils/validation.py
import jsonschema
import logging
import json
from config import DEBUG_MODE
logging.basicConfig(filename="logs/debug.log", level=logging.DEBUG if DEBUG_MODE else logging.INFO)
with open("schemas/function_schema.json") as f:
function_schema = json.load(f)
def validate_function(function_data):
try:
jsonschema.validate(instance=function_data, schema=function_schema)
logging.debug(f"Validation passed: {function_data.get('name', 'unknown')}")
return True
except jsonschema.ValidationError as e:
logging.error(f"Validation error: {e}")
return False
# utils/db_utils.py
import sqlite3
from pathlib import Path
db_path = Path("binary_decompilation.db")
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Functions table
cursor.execute('''
CREATE TABLE IF NOT EXISTS functions (
function_id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE,
address TEXT,
entry_point TEXT,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Decompilation Output table
cursor.execute('''
CREATE TABLE IF NOT EXISTS decompilation_output (
function_id INTEGER,
assembly_code TEXT,
pseudo_c_code TEXT,
last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (function_id) REFERENCES functions(function_id)
)
''')
# C++ Generation table
cursor.execute('''
CREATE TABLE IF NOT EXISTS cpp_generation (
function_id INTEGER,
cpp_code TEXT,
warnings TEXT,
feedback TEXT,
validation_status TEXT,
last_generated TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (function_id) REFERENCES functions(function_id)
)
''')
# Verification and Feedback table
cursor.execute('''
CREATE TABLE IF NOT EXISTS verification_feedback (
function_id INTEGER,
discrepancies TEXT,
refined_prompt TEXT,
verification_status TEXT,
last_verified TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (function_id) REFERENCES functions(function_id)
)
''')
conn.commit()
# Project Metadata table
cursor.execute('''
CREATE TABLE IF NOT EXISTS project_metadata (
project_id INTEGER PRIMARY KEY AUTOINCREMENT,
project_file_name TEXT,
last_modified TEXT,
readonly BOOLEAN,
program_name TEXT,
language_id TEXT,
compiler_id TEXT,
processor TEXT,
endian TEXT,
address_size INTEGER,
min_address TEXT,
max_address TEXT,
num_bytes INTEGER,
num_memory_blocks INTEGER,
num_instructions INTEGER,
num_defined_data INTEGER,
num_functions INTEGER,
num_symbols INTEGER,
num_data_types INTEGER,
analyzed BOOLEAN,
created_with_ghidra_version TEXT,
file_type TEXT,
file_location TEXT,
elf_original_image_base TEXT,
relocatable BOOLEAN,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Missing Libraries table
cursor.execute('''
CREATE TABLE IF NOT EXISTS missing_libraries (
library_id INTEGER PRIMARY KEY AUTOINCREMENT,
project_id INTEGER,
library_name TEXT,
last_checked TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (project_id) REFERENCES project_metadata(project_id)
)
''')
conn.commit()
def save_project_metadata(metadata):
fields = ', '.join(metadata.keys())
placeholders = ', '.join(['?'] * len(metadata))
values = tuple(metadata.values())
cursor.execute(f'''
INSERT INTO project_metadata ({fields})
VALUES ({placeholders})
''', values)
conn.commit()
def save_missing_library(project_id, library_name):
cursor.execute('''
INSERT INTO missing_libraries (project_id, library_name)
VALUES (?, ?)
''', (project_id, library_name))
conn.commit()
def save_to_database(function_id, data, table):
try:
fields = ', '.join(data.keys())
placeholders = ', '.join(['?'] * len(data))
values = tuple(data.values())
cursor.execute(f'''
INSERT OR REPLACE INTO {table} ({fields})
VALUES ({placeholders})
''', values)
conn.commit()
except sqlite3.Error as e:
print(f"Database error: {e}")
def load_function_from_database(function_name):
cursor.execute('SELECT * FROM functions WHERE name = ?', (function_name,))
return cursor.fetchone()
def close_database():
conn.close()
# context manager for database operations
from contextlib import contextmanager
@contextmanager
def get_db_cursor():
conn = sqlite3.connect(db_path)
try:
yield conn.cursor()
finally:
conn.commit()
conn.close()
# utils/feedback_loop.py
import logging
from utils.db_utils import save_to_database
from config import DEBUG_MODE
logging.basicConfig(filename="logs/debug.log", level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def log_discrepancies(function_id, discrepancy_details):
logging.debug(f"Logging discrepancy for function_id '{function_id}': {discrepancy_details}")
save_to_database(function_id, {"discrepancies": str(discrepancy_details)}, "verification_feedback")
def refine_prompts(discrepancies):
if not discrepancies:
logging.debug("No discrepancies found; no prompt refinement necessary.")
return
for func_id, details in discrepancies:
logging.debug(f"Refining prompts based on discrepancy for function_id '{func_id}'")
# Adjustments to prompt structure or content based on discrepancy type
if "type_mismatch" in details:
prompt_modifier = "Ensure strict type compliance in generated C++."
elif "missing_code" in details:
prompt_modifier = "Double-check all control flows and branches."
refined_prompt = {
"role": "user",
"content": f"Refine the C++ output based on these observations:\n{details}\n{prompt_modifier}"
}
save_to_database(func_id, {"refined_prompt": str(refined_prompt)}, "verification_feedback")
logging.debug(f"Updated prompt for function_id '{func_id}': {refined_prompt}")
def feedback_loop(decompiled_functions, verified_functions):
discrepancies = []
for func_id, details in decompiled_functions.items():
if func_id not in verified_functions or verified_functions[func_id]["cpp_code"] != details["cpp_code"]:
discrepancy_details = {"decompiled": details, "verified": verified_functions.get(func_id)}
discrepancies.append((func_id, discrepancy_details))
log_discrepancies(func_id, discrepancy_details)
refine_prompts(discrepancies)
# utils/project_info.py
import os
import logging
import pyhidra
from utils.db_utils import save_project_metadata, save_missing_library
logging.basicConfig(filename="logs/project_info.log", level=logging.INFO)
def gather_project_metadata(binary_path):
with pyhidra.open_program(binary_path) as flat_api:
program = flat_api.getCurrentProgram()
metadata = {
"project_file_name": program.getDomainFile().getName(),
"last_modified": program.getModificationDate().toString(),
"readonly": program.isReadonly(),
"program_name": program.getName(),
"language_id": program.getLanguageID().toString(),
"compiler_id": program.getCompilerSpec().getCompilerSpecID().toString(),
"processor": program.getLanguage().getProcessor().toString(),
"endian": program.getLanguage().isBigEndian(),
"address_size": program.getDefaultPointerSize(),
"min_address": program.getMinAddress().toString(),
"max_address": program.getMaxAddress().toString(),
"num_bytes": program.getMemory().getNumAddresses(),
"num_memory_blocks": program.getMemoryBlockCount(),
"num_instructions": program.getListing().getNumInstructions(),
"num_defined_data": program.getListing().getNumDefinedData(),
"num_functions": program.getFunctionManager().getFunctionCount(),
"num_symbols": program.getSymbolTable().getNumSymbols(),
"num_data_types": len(program.getDataTypeManager().getAllDataTypes()),
"analyzed": program.isAnalyzed(),
"created_with_ghidra_version": program.getVersion(),
"file_type": program.getExecutableFormat(),
"file_location": program.getExecutablePath(),
"elf_original_image_base": program.getExecutableBase(),
"relocatable": program.isRelocatable()
}
save_project_metadata(metadata)
# Log and save required libraries
required_libs = program.getMemory().getExternalLibraries()
for lib in required_libs:
if not os.path.exists(lib):
logging.warning(f"Missing library: {lib}")
save_missing_library(lib)
logging.info("Project metadata and library information gathered and saved.")
# utils/cpp_generator.py
import logging
from config import LLM_MODEL, DEBUG_MODE, MAX_CONTEXT_LENGTH, MAX_TOKENS, MODEL_FUNCTION_CALL_SETTINGS, LOG_FILE
from litellm import completion
logging.basicConfig(filename=LOG_FILE, level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def generate_cpp_code(pseudo_c_code, function_name, symbol_names="", comments=""):
try:
# Include symbols and comments in the prompt for context-awareness
prompt_content = f"Convert the following pseudo-C code to structured C++:\n{pseudo_c_code}"
if symbol_names or comments:
prompt_content += f"\n\nSymbols:\n{symbol_names}\nComments:\n{comments}"
cpp_code_response = completion(
model=LLM_MODEL,
messages=[{
"role": "user",
"content": prompt_content
}],
format="json",
max_context_length=MAX_CONTEXT_LENGTH,
max_tokens=MAX_TOKENS,
function_call=MODEL_FUNCTION_CALL_SETTINGS
)
# Similarity check and feedback trigger
cpp_code = cpp_code_response.get('output', {}).get('cpp_code', '')
if not check_similarity(pseudo_c_code, cpp_code):
logging.warning(f"Similarity check failed for '{function_name}'. Triggering feedback loop.")
feedback_loop({function_name: function_metadata})
return cpp_code
except Exception as e:
logging.error(f"Error generating C++ code for function '{function_name}': {str(e)}")
return None
# utils/generate_file_structure.py
import os
import logging
from config import (DATABASE_PATH, CPP_OUTPUT_DIR, CMAKE_MINIMUM_VERSION, TARGET_NAME, CXX_STANDARD,
ADDITIONAL_LIBRARIES, LLVM_PATH, INCLUDE_DIRECTORIES, DEBUG_MODE, LOG_FILE)
logging.basicConfig(filename=LOG_FILE, level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def generate_files_from_db():
# Connect to database and fetch functions
with sqlite3.connect(DATABASE_PATH) as conn:
cursor = conn.cursor()
project_metadata = gather_project_metadata()
cursor.execute("SELECT function_id, name, cpp_code, offset, signature FROM cpp_generation WHERE validation_status = 'verified'")
functions = cursor.fetchall()
header_content = {}
source_content = {}
for func_id, func_name, cpp_code, offset, signature in functions:
header_file, source_file = determine_file_structure(func_name, project_metadata)
if header_file not in header_content:
header_content[header_file] = ""
header_content[header_file] += f"{signature};\n"
if source_file not in source_content:
source_content[source_file] = ""
source_content[source_file] += cpp_code + "\n"
# Write header and source files
for filename, content in header_content.items():
with open(os.path.join(CPP_OUTPUT_DIR, "include", filename), "w") as f:
f.write("#pragma once\n\n" + content)
for filename, content in source_content.items():
with open(os.path.join(CPP_OUTPUT_DIR, "src", filename), "w") as f:
f.write(content)
generate_cmake_file(project_metadata)
def determine_file_structure(func_name, project_metadata):
module = project_metadata.get("modules", {}).get(func_name)
namespace = project_metadata.get("namespaces", {}).get(func_name)
if module:
module_dir = os.path.join(CPP_OUTPUT_DIR, "src", module)
os.makedirs(module_dir, exist_ok=True)
header_file = os.path.join(module, f"{func_name}.h")
source_file = os.path.join(module, f"{func_name}.cpp")
elif namespace:
namespace_dir = os.path.join(CPP_OUTPUT_DIR, "src", namespace)
os.makedirs(namespace_dir, exist_ok=True)
header_file = os.path.join(namespace, f"{func_name}.h")
source_file = os.path.join(namespace, f"{func_name}.cpp")
else:
header_file = f"{func_name}.h"
source_file = f"{func_name}.cpp"
return header_file, source_file
def generate_cmake_file(project_metadata):
# Build CMake content
library_includes = "\n".join([f"target_link_libraries({TARGET_NAME} {lib})" for lib in ADDITIONAL_LIBRARIES])
include_directories = "\n".join([f"include_directories({dir})" for dir in INCLUDE_DIRECTORIES])
cmake_content = f"""
cmake_minimum_required(VERSION {CMAKE_MINIMUM_VERSION})
project({TARGET_NAME})
set(CMAKE_CXX_STANDARD {CXX_STANDARD})
{include_directories}
# Add sources
file(GLOB SOURCES "src/**/*.cpp")
# Define executable
add_executable({TARGET_NAME} ${{SOURCES}})
# Link libraries
{library_includes}
# LLVM configuration (if needed)
if (EXISTS "{LLVM_PATH}")
find_package(LLVM REQUIRED PATHS "{LLVM_PATH}")
target_include_directories({TARGET_NAME} PRIVATE ${{LLVM_INCLUDE_DIRS}})
target_link_libraries({TARGET_NAME} ${{LLVM_LIBS}})
add_definitions(${{LLVM_DEFINITIONS}})
endif()
"""
with open(os.path.join(CPP_OUTPUT_DIR, "CMakeLists.txt"), "w") as cmake_file:
cmake_file.write(cmake_content)
function_schema.json:
{
"type": "object",
"properties": {
"name": {"type": "string"},
"address": {"type": "string"},
"offset": {"type": "integer"},
"instructions": {
"type": "array",
"items": {"type": "string"}
},
"pseudo_code": {"type": "string"},
"cpp_code": {"type": "string"}
},
"required": ["name", "address", "instructions", "pseudo_code", "cpp_code"]
}
```
Assistant: The provided Python code is part of an automated system designed for binary decompilation, validation, verification, and C++ code generation. Here's a high-level overview of what each component does:
1. **Main Program (`main.py`)**:
- Initializes by gathering metadata about a binary file (`Civ5XP`).
- Creates an output directory for C++ files if it doesn't exist.
- Executes the `decompile_functions` function to decompile the binary file and store its functions.
- Validates each decompiled function against a predefined schema.
- Saves valid functions to a database.
- Verifies decompiled functions by comparing with the binary code.
- Executes a feedback loop to refine decompilation and verification results.
- Generates C++ source and header files from the database and creates a CMake build configuration.
- Closes the database connection.
2. **Configuration (`config.py`)**:
- Contains configuration settings such as paths for binary and output files, logging, and C++ generation settings.
- Configures the language model for generating C++ code from pseudo-C code.
3. **Decompilation Utilities (`decompilation.py`)**:
- Uses the `pyhidra` library to open and analyze a binary file.
- Decompiles functions from this binary and attempts to convert them into pseudo-C code.
- Utilizes `generate_cpp_code` to translate pseudo-C into C++.
- Validates each function's metadata and stores it in a database if valid.
- Calls the `verify_function` method to ensure the generated C++ code matches the original binary’s behavior.
4. **Verification (`verification.py`)**:
- Uses `angr` for static binary analysis to create a control flow graph from the binary.
- Compares the generated LLVM IR from the C++ code with the analyzed binary.
- Validates that function addresses from LLVM IR match those found in the binary.
5. **Validation (`validation.py`)**:
- Validates function metadata using a JSON schema to ensure it matches expected data structures.
- Logs any validation errors.
6. **Database Utilities (`db_utils.py`)**:
- Defines and manages several SQLite tables to store function information, decompilation outputs, C++ generation data, and verification feedback.
- Provides utilities to save, fetch, and update database records.
7. **Feedback Loop (`feedback_loop.py`)**:
- Logs discrepancies found during verification, updates database with feedback, and refines the input prompts.
8. **Project Info (`project_info.py`)**:
- Collects metadata about the project using `pyhidra`, such as function counts, memory blocks, symbols, etc.
- Also checks for missing required libraries and logs this information.
9. **C++ Code Generation (`cpp_generator.py`)**:
- Uses a language model (`ollama/qwen2.5:14b`) to translate pseudo-C code into structured C++.
- Conducts a consistency check between pseudo-C and generated C++ and triggers feedback if necessary.
10. **File Structuring (`generate_file_structure.py`)**:
- Retrieves verified functions from the database and organizes them into header and source files.
- Generates a CMake file to facilitate building the decompiled C++ project.
Overall, this code appears to be an integrated system for decompiling a binary, generating equivalent C++ code, and ensuring its correctness through validation and verification.
User: similarity_analysis.py # New: Calculates similarity ratios and graph-based comparisons for validation
similarity_analysis.py utility for additional validation techniques, such as similarity ratios and graph-based comparisons
Implement the new similarity_analysis.py within the project to ensure the automatic bruteforcing reverse engineering to gain the C++ code improve in the loop and doesn't degrade or loop for no benefit.
```
# main.py
import os
from config import BINARY_PATH, CPP_OUTPUT_DIR
from utils.project_info import gather_project_metadata
from utils.decompilation import decompile_functions
from utils.validation import validate_function
from utils.verification import verify_function
from utils.db_utils import save_to_database, close_database
from utils.generate_file_structure import generate_files_from_db
def main():
gather_project_metadata(BINARY_PATH)
os.makedirs(CPP_OUTPUT_DIR, exist_ok=True)
decompiled_functions = decompile_functions(BINARY_PATH)
for func_name, func_data in decompiled_functions.items():
if validate_function(func_data):
save_to_database(func_name, func_data['offset'], func_data['assembly_code'], func_data['decompiled_code'], "")
verified_functions = verify_function(BINARY_PATH, decompiled_functions)
feedback_loop(decompiled_functions, verified_functions)
generate_files_from_db()
close_database()
if __name__ == "__main__":
main()
# config.py
# Base URL for the Ollama API running on WSL2
OLLAMA_BASE_URL = "http://172.31.72.252:11434"
# Model to use in Ollama
LLM_MODEL = "ollama/qwen2.5:14b"
# Debug mode toggle
DEBUG_MODE = True
# Paths for project files and directories
BINARY_PATH = r"Civ5XP" # Path to the binary file for decompilation
CPP_OUTPUT_DIR = "output" # Directory for C++ output files
LOG_FILE = "logs/debug.log" # Log file path
DATABASE_PATH = "binary_decompilation.db" # Database file path
# Decompilation and C++ generation settings
MAX_CONTEXT_LENGTH = 128000 # Max context length for LLM input
MAX_TOKENS = 8192 # Max tokens for LLM output
CXX_STANDARD = 17 # C++ standard for generated code
# Function call model configuration for LLM
MODEL_FUNCTION_CALL_SETTINGS = {
"name": "generate_cpp",
"parameters": [
{"name": "cpp_code", "type": "string"},
{"name": "warnings", "type": "list"},
{"name": "feedback", "type": "string"}
]
}
# CMake file generation options
CMAKE_MINIMUM_VERSION = "3.10"
TARGET_NAME = "DecompiledProject" # Target name in CMakeLists.txt
ADDITIONAL_LIBRARIES = ["libA", "libB"] # Additional libraries to link in CMakeLists.txt
LLVM_PATH = "/usr/local/llvm" # Adjust to your actual LLVM path if needed
INCLUDE_DIRECTORIES = ["include", "/path/to/other/includes"]
```
```
# utils/decompilation.py
from utils.verification import verify_function, compile_cpp_to_llvm_ir
from utils.validation import validate_function
from utils.feedback_loop import feedback_loop
def decompile_functions(binary_file=BINARY_PATH):
project_metadata = gather_project_metadata(binary_file)
function_summaries = {}
with pyhidra.open_program(binary_file) as flat_api:
program = flat_api.getCurrentProgram()
listing = program.getListing()
decompiler = DecompInterface()
decompiler.openProgram(program)
for function in listing.getFunctions(True):
function_name = function.getName()
function_offset = function.getEntryPoint().getOffset()
db_entry = load_function_from_database(function_name)
if db_entry:
logging.debug(f"Function '{function_name}' already in database.")
continue
try:
results = decompiler.decompileFunction(function, 0, TaskMonitor.DUMMY)
if results and results.decompiledFunction is not None:
decompiled_code = results.getDecompiledFunction().getC()
assembly_code = str(function.getBody())
cpp_code = generate_cpp_code(decompiled_code, function_name)
# Decompile function and collect metadata
function_metadata = {
"name": function_name,
"offset": function_offset,
"assembly_code": assembly_code,
"decompiled_code": decompiled_code,
"symbol_names": [symbol.getName() for symbol in function.getSymbols()],
"comments": extract_comments_from_function(function),
"cpp_code": cpp_code
}
# Store to database after validation
if validate_function(function_metadata):
save_to_database(function_name, function_metadata, 'decompilation_output')
function_summaries[function_name] = function_metadata
# Verification step - LLVM IR and Control Flow Graph matching
llvm_ir = compile_cpp_to_llvm_ir(cpp_code)
verification_result = verify_function(binary_file, {function_name: function_metadata})
if not verification_result:
logging.warning(f"Verification failed for '{function_name}'. Triggering feedback loop.")
feedback_loop({function_name: function_metadata}, function_summaries)
else:
logging.warning(f"Validation failed for '{function_name}'.")
else:
logging.error(f"Failed decompiling '{function_name}'.")
except Exception as e:
logging.error(f"Error processing '{function_name}': {str(e)}")
close_database()
return function_summaries
```
```
# utils/verification.py
import angr
import llvmlite.binding as llvm
import logging
import subprocess
import tempfile
from config import DEBUG_MODE
from utils.db_utils import load_project_metadata
logging.basicConfig(filename="logs/debug.log", level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def verify_function(binary_path, functions_data):
project_metadata = load_project_metadata()
min_address = int(project_metadata.get("min_address", "0"), 16)
max_address = int(project_metadata.get("max_address", "FFFFFFFF"), 16)
for func_name, func_data in functions_data.items():
cpp_code = func_data.get("cpp_code")
if cpp_code:
llvm_ir = compile_cpp_to_llvm_ir(cpp_code)
if llvm_ir:
result = compare_with_binary(binary_path, func_name, llvm_ir, min_address, max_address)
if result:
logging.info(f"Verification passed for '{func_name}'.")
else:
logging.warning(f"Verification failed for '{func_name}'.")
def compile_cpp_to_llvm_ir(cpp_code):
try:
with tempfile.NamedTemporaryFile(suffix=".cpp", delete=False) as cpp_file:
cpp_file.write(cpp_code.encode())
cpp_filename = cpp_file.name
llvm_filename = cpp_filename.replace(".cpp", ".ll")
subprocess.run(["clang", "-emit-llvm", "-S", cpp_filename, "-o", llvm_filename], check=True)
with open(llvm_filename, "r") as llvm_ir_file:
llvm_ir = llvm_ir_file.read()
return llvm_ir
except subprocess.CalledProcessError as e:
logging.error(f"Clang compilation failed: {str(e)}")
return None
except Exception as e:
logging.error(f"Error compiling C++ to LLVM IR: {str(e)}")
return None
def compare_with_binary(binary_path, func_name, llvm_ir, min_address, max_address):
try:
project = angr.Project(binary_path, auto_load_libs=False)
cfg = project.analyses.CFGFast()
llvm_func_addresses = extract_function_addresses_from_llvm(llvm_ir, func_name)
for address in llvm_func_addresses:
if not (min_address <= address <= max_address):
logging.warning(f"Address {address} for '{func_name}' out of valid range.")
continue
binary_func = cfg.kb.functions.get_by_addr(address)
if binary_func is None:
logging.warning(f"Function '{func_name}' at address {address} not found in binary.")
return False
return True
except Exception as e:
logging.error(f"Error comparing LLVM IR with binary: {str(e)}")
return False
def extract_function_addresses_from_llvm(llvm_ir, func_name):
addresses = []
for line in llvm_ir.splitlines():
if func_name in line and "define" in line:
# Use a more accurate regex or parser for proper extraction
# Assuming format like `@func_name = external addrspace(0) constant i32 0x...`
address_match = re.search(r'0x[0-9A-Fa-f]+', line)
if address_match:
address = int(address_match.group(0), 16)
addresses.append(address)
return addresses
```
```
# utils/validation.py
import jsonschema
import logging
import json
from config import DEBUG_MODE
logging.basicConfig(filename="logs/debug.log", level=logging.DEBUG if DEBUG_MODE else logging.INFO)
with open("schemas/function_schema.json") as f:
function_schema = json.load(f)
def validate_function(function_data):
try:
jsonschema.validate(instance=function_data, schema=function_schema)
logging.debug(f"Validation passed: {function_data.get('name', 'unknown')}")
return True
except jsonschema.ValidationError as e:
logging.error(f"Validation error: {e}")
return False
```
```
# utils/db_utils.py
import sqlite3
from pathlib import Path
db_path = Path("binary_decompilation.db")
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Functions table
cursor.execute('''
CREATE TABLE IF NOT EXISTS functions (
function_id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE,
address TEXT,
entry_point TEXT,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Decompilation Output table
cursor.execute('''
CREATE TABLE IF NOT EXISTS decompilation_output (
function_id INTEGER,
assembly_code TEXT,
pseudo_c_code TEXT,
last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (function_id) REFERENCES functions(function_id)
)
''')
# C++ Generation table
cursor.execute('''
CREATE TABLE IF NOT EXISTS cpp_generation (
function_id INTEGER,
cpp_code TEXT,
warnings TEXT,
feedback TEXT,
validation_status TEXT,
last_generated TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (function_id) REFERENCES functions(function_id)
)
''')
# Verification and Feedback table
cursor.execute('''
CREATE TABLE IF NOT EXISTS verification_feedback (
function_id INTEGER,
discrepancies TEXT,
refined_prompt TEXT,
verification_status TEXT,
last_verified TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (function_id) REFERENCES functions(function_id)
)
''')
conn.commit()
# Project Metadata table
cursor.execute('''
CREATE TABLE IF NOT EXISTS project_metadata (
project_id INTEGER PRIMARY KEY AUTOINCREMENT,
project_file_name TEXT,
last_modified TEXT,
readonly BOOLEAN,
program_name TEXT,
language_id TEXT,
compiler_id TEXT,
processor TEXT,
endian TEXT,
address_size INTEGER,
min_address TEXT,
max_address TEXT,
num_bytes INTEGER,
num_memory_blocks INTEGER,
num_instructions INTEGER,
num_defined_data INTEGER,
num_functions INTEGER,
num_symbols INTEGER,
num_data_types INTEGER,
analyzed BOOLEAN,
created_with_ghidra_version TEXT,
file_type TEXT,
file_location TEXT,
elf_original_image_base TEXT,
relocatable BOOLEAN,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Missing Libraries table
cursor.execute('''
CREATE TABLE IF NOT EXISTS missing_libraries (
library_id INTEGER PRIMARY KEY AUTOINCREMENT,
project_id INTEGER,
library_name TEXT,
last_checked TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (project_id) REFERENCES project_metadata(project_id)
)
''')
conn.commit()
def save_project_metadata(metadata):
fields = ', '.join(metadata.keys())
placeholders = ', '.join(['?'] * len(metadata))
values = tuple(metadata.values())
cursor.execute(f'''
INSERT INTO project_metadata ({fields})
VALUES ({placeholders})
''', values)
conn.commit()
def save_missing_library(project_id, library_name):
cursor.execute('''
INSERT INTO missing_libraries (project_id, library_name)
VALUES (?, ?)
''', (project_id, library_name))
conn.commit()
def save_to_database(function_id, data, table):
try:
fields = ', '.join(data.keys())
placeholders = ', '.join(['?'] * len(data))
values = tuple(data.values())
cursor.execute(f'''
INSERT OR REPLACE INTO {table} ({fields})
VALUES ({placeholders})
''', values)
conn.commit()
except sqlite3.Error as e:
print(f"Database error: {e}")
def load_function_from_database(function_name):
cursor.execute('SELECT * FROM functions WHERE name = ?', (function_name,))
return cursor.fetchone()
def close_database():
conn.close()
# context manager for database operations
from contextlib import contextmanager
@contextmanager
def get_db_cursor():
conn = sqlite3.connect(db_path)
try:
yield conn.cursor()
finally:
conn.commit()
conn.close()
```
```
# utils/feedback_loop.py
import logging
from utils.db_utils import save_to_database
from config import DEBUG_MODE
logging.basicConfig(filename="logs/debug.log", level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def log_discrepancies(function_id, discrepancy_details):
logging.debug(f"Logging discrepancy for function_id '{function_id}': {discrepancy_details}")
save_to_database(function_id, {"discrepancies": str(discrepancy_details)}, "verification_feedback")
def refine_prompts(discrepancies):
if not discrepancies:
logging.debug("No discrepancies found; no prompt refinement necessary.")
return
for func_id, details in discrepancies:
logging.debug(f"Refining prompts based on discrepancy for function_id '{func_id}'")
# Adjustments to prompt structure or content based on discrepancy type
if "type_mismatch" in details:
prompt_modifier = "Ensure strict type compliance in generated C++."
elif "missing_code" in details:
prompt_modifier = "Double-check all control flows and branches."
refined_prompt = {
"role": "user",
"content": f"Refine the C++ output based on these observations:\n{details}\n{prompt_modifier}"
}
save_to_database(func_id, {"refined_prompt": str(refined_prompt)}, "verification_feedback")
logging.debug(f"Updated prompt for function_id '{func_id}': {refined_prompt}")
def feedback_loop(decompiled_functions, verified_functions):
discrepancies = []
for func_id, details in decompiled_functions.items():
if func_id not in verified_functions or verified_functions[func_id]["cpp_code"] != details["cpp_code"]:
discrepancy_details = {"decompiled": details, "verified": verified_functions.get(func_id)}
discrepancies.append((func_id, discrepancy_details))
log_discrepancies(func_id, discrepancy_details)
refine_prompts(discrepancies)
```
```
# utils/project_info.py
import os
import logging
import pyhidra
from utils.db_utils import save_project_metadata, save_missing_library
logging.basicConfig(filename="logs/project_info.log", level=logging.INFO)
def gather_project_metadata(binary_path):
with pyhidra.open_program(binary_path) as flat_api:
program = flat_api.getCurrentProgram()
metadata = {
"project_file_name": program.getDomainFile().getName(),
"last_modified": program.getModificationDate().toString(),
"readonly": program.isReadonly(),
"program_name": program.getName(),
"language_id": program.getLanguageID().toString(),
"compiler_id": program.getCompilerSpec().getCompilerSpecID().toString(),
"processor": program.getLanguage().getProcessor().toString(),
"endian": program.getLanguage().isBigEndian(),
"address_size": program.getDefaultPointerSize(),
"min_address": program.getMinAddress().toString(),
"max_address": program.getMaxAddress().toString(),
"num_bytes": program.getMemory().getNumAddresses(),
"num_memory_blocks": program.getMemoryBlockCount(),
"num_instructions": program.getListing().getNumInstructions(),
"num_defined_data": program.getListing().getNumDefinedData(),
"num_functions": program.getFunctionManager().getFunctionCount(),
"num_symbols": program.getSymbolTable().getNumSymbols(),
"num_data_types": len(program.getDataTypeManager().getAllDataTypes()),
"analyzed": program.isAnalyzed(),
"created_with_ghidra_version": program.getVersion(),
"file_type": program.getExecutableFormat(),
"file_location": program.getExecutablePath(),
"elf_original_image_base": program.getExecutableBase(),
"relocatable": program.isRelocatable()
}
save_project_metadata(metadata)
# Log and save required libraries
required_libs = program.getMemory().getExternalLibraries()
for lib in required_libs:
if not os.path.exists(lib):
logging.warning(f"Missing library: {lib}")
save_missing_library(lib)
logging.info("Project metadata and library information gathered and saved.")
```
```
# utils/cpp_generator.py
import logging
from config import LLM_MODEL, DEBUG_MODE, MAX_CONTEXT_LENGTH, MAX_TOKENS, MODEL_FUNCTION_CALL_SETTINGS, LOG_FILE
from litellm import completion
logging.basicConfig(filename=LOG_FILE, level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def generate_cpp_code(pseudo_c_code, function_name, symbol_names="", comments=""):
try:
# Include symbols and comments in the prompt for context-awareness
prompt_content = f"Convert the following pseudo-C code to structured C++:\n{pseudo_c_code}"
if symbol_names or comments:
prompt_content += f"\n\nSymbols:\n{symbol_names}\nComments:\n{comments}"
cpp_code_response = completion(
model=LLM_MODEL,
messages=[{
"role": "user",
"content": prompt_content
}],
format="json",
max_context_length=MAX_CONTEXT_LENGTH,
max_tokens=MAX_TOKENS,
function_call=MODEL_FUNCTION_CALL_SETTINGS
)
# Similarity check and feedback trigger
cpp_code = cpp_code_response.get('output', {}).get('cpp_code', '')
if not check_similarity(pseudo_c_code, cpp_code):
logging.warning(f"Similarity check failed for '{function_name}'. Triggering feedback loop.")
feedback_loop({function_name: function_metadata})
return cpp_code
except Exception as e:
logging.error(f"Error generating C++ code for function '{function_name}': {str(e)}")
return None
```
```
# utils/generate_file_structure.py
import os
import logging
from config import (DATABASE_PATH, CPP_OUTPUT_DIR, CMAKE_MINIMUM_VERSION, TARGET_NAME, CXX_STANDARD,
ADDITIONAL_LIBRARIES, LLVM_PATH, INCLUDE_DIRECTORIES, DEBUG_MODE, LOG_FILE)
logging.basicConfig(filename=LOG_FILE, level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def generate_files_from_db():
# Connect to database and fetch functions
with sqlite3.connect(DATABASE_PATH) as conn:
cursor = conn.cursor()
project_metadata = gather_project_metadata()
cursor.execute("SELECT function_id, name, cpp_code, offset, signature FROM cpp_generation WHERE validation_status = 'verified'")
functions = cursor.fetchall()
header_content = {}
source_content = {}
for func_id, func_name, cpp_code, offset, signature in functions:
header_file, source_file = determine_file_structure(func_name, project_metadata)
if header_file not in header_content:
header_content[header_file] = ""
header_content[header_file] += f"{signature};\n"
if source_file not in source_content:
source_content[source_file] = ""
source_content[source_file] += cpp_code + "\n"
# Write header and source files
for filename, content in header_content.items():
with open(os.path.join(CPP_OUTPUT_DIR, "include", filename), "w") as f:
f.write("#pragma once\n\n" + content)
for filename, content in source_content.items():
with open(os.path.join(CPP_OUTPUT_DIR, "src", filename), "w") as f:
f.write(content)
generate_cmake_file(project_metadata)
def determine_file_structure(func_name, project_metadata):
module = project_metadata.get("modules", {}).get(func_name)
namespace = project_metadata.get("namespaces", {}).get(func_name)
if module:
module_dir = os.path.join(CPP_OUTPUT_DIR, "src", module)
os.makedirs(module_dir, exist_ok=True)
header_file = os.path.join(module, f"{func_name}.h")
source_file = os.path.join(module, f"{func_name}.cpp")
elif namespace:
namespace_dir = os.path.join(CPP_OUTPUT_DIR, "src", namespace)
os.makedirs(namespace_dir, exist_ok=True)
header_file = os.path.join(namespace, f"{func_name}.h")
source_file = os.path.join(namespace, f"{func_name}.cpp")
else:
header_file = f"{func_name}.h"
source_file = f"{func_name}.cpp"
return header_file, source_file
def generate_cmake_file(project_metadata):
# Build CMake content
library_includes = "\n".join([f"target_link_libraries({TARGET_NAME} {lib})" for lib in ADDITIONAL_LIBRARIES])
include_directories = "\n".join([f"include_directories({dir})" for dir in INCLUDE_DIRECTORIES])
cmake_content = f"""
cmake_minimum_required(VERSION {CMAKE_MINIMUM_VERSION})
project({TARGET_NAME})
set(CMAKE_CXX_STANDARD {CXX_STANDARD})
{include_directories}
# Add sources
file(GLOB SOURCES "src/**/*.cpp")
# Define executable
add_executable({TARGET_NAME} ${{SOURCES}})
# Link libraries
{library_includes}
# LLVM configuration (if needed)
if (EXISTS "{LLVM_PATH}")
find_package(LLVM REQUIRED PATHS "{LLVM_PATH}")
target_include_directories({TARGET_NAME} PRIVATE ${{LLVM_INCLUDE_DIRS}})
target_link_libraries({TARGET_NAME} ${{LLVM_LIBS}})
add_definitions(${{LLVM_DEFINITIONS}})
endif()
"""
with open(os.path.join(CPP_OUTPUT_DIR, "CMakeLists.txt"), "w") as cmake_file:
cmake_file.write(cmake_content)
```
```
function_schema.json:
{
"type": "object",
"properties": {
"name": {"type": "string"},
"address": {"type": "string"},
"offset": {"type": "integer"},
"instructions": {
"type": "array",
"items": {"type": "string"}
},
"pseudo_code": {"type": "string"},
"cpp_code": {"type": "string"}
},
"required": ["name", "address", "instructions", "pseudo_code", "cpp_code"]
}
```
```plaintext
# main.py
import os
from config import BINARY_PATH, CPP_OUTPUT_DIR
from utils.project_info import gather_project_metadata
from utils.decompilation import decompile_functions
from utils.validation import validate_function
from utils.verification import verify_function
from utils.db_utils import save_to_database, close_database
from utils.generate_file_structure import generate_files_from_db
def main():
gather_project_metadata(BINARY_PATH)
os.makedirs(CPP_OUTPUT_DIR, exist_ok=True)
decompiled_functions = decompile_functions(BINARY_PATH)
for func_name, func_data in decompiled_functions.items():
if validate_function(func_data):
save_to_database(func_name, func_data['offset'], func_data['assembly_code'], func_data['decompiled_code'], "")
verified_functions = verify_function(BINARY_PATH, decompiled_functions)
feedback_loop(decompiled_functions, verified_functions)
generate_files_from_db()
close_database()
if __name__ == "__main__":
main()
# config.py
# Base URL for the Ollama API running on WSL2
OLLAMA_BASE_URL = "http://172.31.72.252:11434"
# Model to use in Ollama
LLM_MODEL = "ollama/qwen2.5:14b"
# Debug mode toggle
DEBUG_MODE = True
# Paths for project files and directories
BINARY_PATH = r"Civ5XP" # Path to the binary file for decompilation
CPP_OUTPUT_DIR = "output" # Directory for C++ output files
LOG_FILE = "logs/debug.log" # Log file path
DATABASE_PATH = "binary_decompilation.db" # Database file path
# Decompilation and C++ generation settings
MAX_CONTEXT_LENGTH = 128000 # Max context length for LLM input
MAX_TOKENS = 8192 # Max tokens for LLM output
CXX_STANDARD = 17 # C++ standard for generated code
# Function call model configuration for LLM
MODEL_FUNCTION_CALL_SETTINGS = {
"name": "generate_cpp",
"parameters": [
{"name": "cpp_code", "type": "string"},
{"name": "warnings", "type": "list"},
{"name": "feedback", "type": "string"}
]
}
# CMake file generation options
CMAKE_MINIMUM_VERSION = "3.10"
TARGET_NAME = "DecompiledProject" # Target name in CMakeLists.txt
ADDITIONAL_LIBRARIES = ["libA", "libB"] # Additional libraries to link in CMakeLists.txt
LLVM_PATH = "/usr/local/llvm" # Adjust to your actual LLVM path if needed
INCLUDE_DIRECTORIES = ["include", "/path/to/other/includes"]
# utils/decompilation.py
from utils.verification import verify_function, compile_cpp_to_llvm_ir
from utils.validation import validate_function
from utils.feedback_loop import feedback_loop
def decompile_functions(binary_file=BINARY_PATH):
project_metadata = gather_project_metadata(binary_file)
function_summaries = {}
with pyhidra.open_program(binary_file) as flat_api:
program = flat_api.getCurrentProgram()
listing = program.getListing()
decompiler = DecompInterface()
decompiler.openProgram(program)
for function in listing.getFunctions(True):
function_name = function.getName()
function_offset = function.getEntryPoint().getOffset()
db_entry = load_function_from_database(function_name)
if db_entry:
logging.debug(f"Function '{function_name}' already in database.")
continue
try:
results = decompiler.decompileFunction(function, 0, TaskMonitor.DUMMY)
if results and results.decompiledFunction is not None:
decompiled_code = results.getDecompiledFunction().getC()
assembly_code = str(function.getBody())
cpp_code = generate_cpp_code(decompiled_code, function_name)
# Decompile function and collect metadata
function_metadata = {
"name": function_name,
"offset": function_offset,
"assembly_code": assembly_code,
"decompiled_code": decompiled_code,
"symbol_names": [symbol.getName() for symbol in function.getSymbols()],
"comments": extract_comments_from_function(function),
"cpp_code": cpp_code
}
# Store to database after validation
if validate_function(function_metadata):
save_to_database(function_name, function_metadata, 'decompilation_output')
function_summaries[function_name] = function_metadata
# Verification step - LLVM IR and Control Flow Graph matching
llvm_ir = compile_cpp_to_llvm_ir(cpp_code)
verification_result = verify_function(binary_file, {function_name: function_metadata})
if not verification_result:
logging.warning(f"Verification failed for '{function_name}'. Triggering feedback loop.")
feedback_loop({function_name: function_metadata}, function_summaries)
else:
logging.warning(f"Validation failed for '{function_name}'.")
else:
logging.error(f"Failed decompiling '{function_name}'.")
except Exception as e:
logging.error(f"Error processing '{function_name}': {str(e)}")
close_database()
return function_summaries
# utils/verification.py
import angr
import llvmlite.binding as llvm
import logging
import subprocess
import tempfile
from config import DEBUG_MODE
from utils.db_utils import load_project_metadata
logging.basicConfig(filename="logs/debug.log", level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def verify_function(binary_path, functions_data):
project_metadata = load_project_metadata()
min_address = int(project_metadata.get("min_address", "0"), 16)
max_address = int(project_metadata.get("max_address", "FFFFFFFF"), 16)
for func_name, func_data in functions_data.items():
cpp_code = func_data.get("cpp_code")
if cpp_code:
llvm_ir = compile_cpp_to_llvm_ir(cpp_code)
if llvm_ir:
result = compare_with_binary(binary_path, func_name, llvm_ir, min_address, max_address)
if result:
logging.info(f"Verification passed for '{func_name}'.")
else:
logging.warning(f"Verification failed for '{func_name}'.")
def compile_cpp_to_llvm_ir(cpp_code):
try:
with tempfile.NamedTemporaryFile(suffix=".cpp", delete=False) as cpp_file:
cpp_file.write(cpp_code.encode())
cpp_filename = cpp_file.name
llvm_filename = cpp_filename.replace(".cpp", ".ll")
subprocess.run(["clang", "-emit-llvm", "-S", cpp_filename, "-o", llvm_filename], check=True)
with open(llvm_filename, "r") as llvm_ir_file:
llvm_ir = llvm_ir_file.read()
return llvm_ir
except subprocess.CalledProcessError as e:
logging.error(f"Clang compilation failed: {str(e)}")
return None
except Exception as e:
logging.error(f"Error compiling C++ to LLVM IR: {str(e)}")
return None
def compare_with_binary(binary_path, func_name, llvm_ir, min_address, max_address):
try:
project = angr.Project(binary_path, auto_load_libs=False)
cfg = project.analyses.CFGFast()
llvm_func_addresses = extract_function_addresses_from_llvm(llvm_ir, func_name)
for address in llvm_func_addresses:
if not (min_address <= address <= max_address):
logging.warning(f"Address {address} for '{func_name}' out of valid range.")
continue
binary_func = cfg.kb.functions.get_by_addr(address)
if binary_func is None:
logging.warning(f"Function '{func_name}' at address {address} not found in binary.")
return False
return True
except Exception as e:
logging.error(f"Error comparing LLVM IR with binary: {str(e)}")
return False
def extract_function_addresses_from_llvm(llvm_ir, func_name):
addresses = []
for line in llvm_ir.splitlines():
if func_name in line and "define" in line:
# Use a more accurate regex or parser for proper extraction
# Assuming format like `@func_name = external addrspace(0) constant i32 0x...`
address_match = re.search(r'0x[0-9A-Fa-f]+', line)
if address_match:
address = int(address_match.group(0), 16)
addresses.append(address)
return addresses
# utils/validation.py
import jsonschema
import logging
import json
from config import DEBUG_MODE
logging.basicConfig(filename="logs/debug.log", level=logging.DEBUG if DEBUG_MODE else logging.INFO)
with open("schemas/function_schema.json") as f:
function_schema = json.load(f)
def validate_function(function_data):
try:
jsonschema.validate(instance=function_data, schema=function_schema)
logging.debug(f"Validation passed: {function_data.get('name', 'unknown')}")
return True
except jsonschema.ValidationError as e:
logging.error(f"Validation error: {e}")
return False
# utils/db_utils.py
import sqlite3
from pathlib import Path
db_path = Path("binary_decompilation.db")
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Functions table
cursor.execute('''
CREATE TABLE IF NOT EXISTS functions (
function_id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE,
address TEXT,
entry_point TEXT,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Decompilation Output table
cursor.execute('''
CREATE TABLE IF NOT EXISTS decompilation_output (
function_id INTEGER,
assembly_code TEXT,
pseudo_c_code TEXT,
last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (function_id) REFERENCES functions(function_id)
)
''')
# C++ Generation table
cursor.execute('''
CREATE TABLE IF NOT EXISTS cpp_generation (
function_id INTEGER,
cpp_code TEXT,
warnings TEXT,
feedback TEXT,
validation_status TEXT,
last_generated TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (function_id) REFERENCES functions(function_id)
)
''')
# Verification and Feedback table
cursor.execute('''
CREATE TABLE IF NOT EXISTS verification_feedback (
function_id INTEGER,
discrepancies TEXT,
refined_prompt TEXT,
verification_status TEXT,
last_verified TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (function_id) REFERENCES functions(function_id)
)
''')
conn.commit()
# Project Metadata table
cursor.execute('''
CREATE TABLE IF NOT EXISTS project_metadata (
project_id INTEGER PRIMARY KEY AUTOINCREMENT,
project_file_name TEXT,
last_modified TEXT,
readonly BOOLEAN,
program_name TEXT,
language_id TEXT,
compiler_id TEXT,
processor TEXT,
endian TEXT,
address_size INTEGER,
min_address TEXT,
max_address TEXT,
num_bytes INTEGER,
num_memory_blocks INTEGER,
num_instructions INTEGER,
num_defined_data INTEGER,
num_functions INTEGER,
num_symbols INTEGER,
num_data_types INTEGER,
analyzed BOOLEAN,
created_with_ghidra_version TEXT,
file_type TEXT,
file_location TEXT,
elf_original_image_base TEXT,
relocatable BOOLEAN,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Missing Libraries table
cursor.execute('''
CREATE TABLE IF NOT EXISTS missing_libraries (
library_id INTEGER PRIMARY KEY AUTOINCREMENT,
project_id INTEGER,
library_name TEXT,
last_checked TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (project_id) REFERENCES project_metadata(project_id)
)
''')
conn.commit()
def save_project_metadata(metadata):
fields = ', '.join(metadata.keys())
placeholders = ', '.join(['?'] * len(metadata))
values = tuple(metadata.values())
cursor.execute(f'''
INSERT INTO project_metadata ({fields})
VALUES ({placeholders})
''', values)
conn.commit()
def save_missing_library(project_id, library_name):
cursor.execute('''
INSERT INTO missing_libraries (project_id, library_name)
VALUES (?, ?)
''', (project_id, library_name))
conn.commit()
def save_to_database(function_id, data, table):
try:
fields = ', '.join(data.keys())
placeholders = ', '.join(['?'] * len(data))
values = tuple(data.values())
cursor.execute(f'''
INSERT OR REPLACE INTO {table} ({fields})
VALUES ({placeholders})
''', values)
conn.commit()
except sqlite3.Error as e:
print(f"Database error: {e}")
def load_function_from_database(function_name):
cursor.execute('SELECT * FROM functions WHERE name = ?', (function_name,))
return cursor.fetchone()
def close_database():
conn.close()
# context manager for database operations
from contextlib import contextmanager
@contextmanager
def get_db_cursor():
conn = sqlite3.connect(db_path)
try:
yield conn.cursor()
finally:
conn.commit()
conn.close()
# utils/feedback_loop.py
import logging
from utils.db_utils import save_to_database
from config import DEBUG_MODE
logging.basicConfig(filename="logs/debug.log", level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def log_discrepancies(function_id, discrepancy_details):
logging.debug(f"Logging discrepancy for function_id '{function_id}': {discrepancy_details}")
save_to_database(function_id, {"discrepancies": str(discrepancy_details)}, "verification_feedback")
def refine_prompts(discrepancies):
if not discrepancies:
logging.debug("No discrepancies found; no prompt refinement necessary.")
return
for func_id, details in discrepancies:
logging.debug(f"Refining prompts based on discrepancy for function_id '{func_id}'")
# Adjustments to prompt structure or content based on discrepancy type
if "type_mismatch" in details:
prompt_modifier = "Ensure strict type compliance in generated C++."
elif "missing_code" in details:
prompt_modifier = "Double-check all control flows and branches."
refined_prompt = {
"role": "user",
"content": f"Refine the C++ output based on these observations:\n{details}\n{prompt_modifier}"
}
save_to_database(func_id, {"refined_prompt": str(refined_prompt)}, "verification_feedback")
logging.debug(f"Updated prompt for function_id '{func_id}': {refined_prompt}")
def feedback_loop(decompiled_functions, verified_functions):
discrepancies = []
for func_id, details in decompiled_functions.items():
if func_id not in verified_functions or verified_functions[func_id]["cpp_code"] != details["cpp_code"]:
discrepancy_details = {"decompiled": details, "verified": verified_functions.get(func_id)}
discrepancies.append((func_id, discrepancy_details))
log_discrepancies(func_id, discrepancy_details)
refine_prompts(discrepancies)
# utils/project_info.py
import os
import logging
import pyhidra
from utils.db_utils import save_project_metadata, save_missing_library
logging.basicConfig(filename="logs/project_info.log", level=logging.INFO)
def gather_project_metadata(binary_path):
with pyhidra.open_program(binary_path) as flat_api:
program = flat_api.getCurrentProgram()
metadata = {
"project_file_name": program.getDomainFile().getName(),
"last_modified": program.getModificationDate().toString(),
"readonly": program.isReadonly(),
"program_name": program.getName(),
"language_id": program.getLanguageID().toString(),
"compiler_id": program.getCompilerSpec().getCompilerSpecID().toString(),
"processor": program.getLanguage().getProcessor().toString(),
"endian": program.getLanguage().isBigEndian(),
"address_size": program.getDefaultPointerSize(),
"min_address": program.getMinAddress().toString(),
"max_address": program.getMaxAddress().toString(),
"num_bytes": program.getMemory().getNumAddresses(),
"num_memory_blocks": program.getMemoryBlockCount(),
"num_instructions": program.getListing().getNumInstructions(),
"num_defined_data": program.getListing().getNumDefinedData(),
"num_functions": program.getFunctionManager().getFunctionCount(),
"num_symbols": program.getSymbolTable().getNumSymbols(),
"num_data_types": len(program.getDataTypeManager().getAllDataTypes()),
"analyzed": program.isAnalyzed(),
"created_with_ghidra_version": program.getVersion(),
"file_type": program.getExecutableFormat(),
"file_location": program.getExecutablePath(),
"elf_original_image_base": program.getExecutableBase(),
"relocatable": program.isRelocatable()
}
save_project_metadata(metadata)
# Log and save required libraries
required_libs = program.getMemory().getExternalLibraries()
for lib in required_libs:
if not os.path.exists(lib):
logging.warning(f"Missing library: {lib}")
save_missing_library(lib)
logging.info("Project metadata and library information gathered and saved.")
# utils/cpp_generator.py
import logging
from config import LLM_MODEL, DEBUG_MODE, MAX_CONTEXT_LENGTH, MAX_TOKENS, MODEL_FUNCTION_CALL_SETTINGS, LOG_FILE
from litellm import completion
logging.basicConfig(filename=LOG_FILE, level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def generate_cpp_code(pseudo_c_code, function_name, symbol_names="", comments=""):
try:
# Include symbols and comments in the prompt for context-awareness
prompt_content = f"Convert the following pseudo-C code to structured C++:\n{pseudo_c_code}"
if symbol_names or comments:
prompt_content += f"\n\nSymbols:\n{symbol_names}\nComments:\n{comments}"
cpp_code_response = completion(
model=LLM_MODEL,
messages=[{
"role": "user",
"content": prompt_content
}],
format="json",
max_context_length=MAX_CONTEXT_LENGTH,
max_tokens=MAX_TOKENS,
function_call=MODEL_FUNCTION_CALL_SETTINGS
)
# Similarity check and feedback trigger
cpp_code = cpp_code_response.get('output', {}).get('cpp_code', '')
if not check_similarity(pseudo_c_code, cpp_code):
logging.warning(f"Similarity check failed for '{function_name}'. Triggering feedback loop.")
feedback_loop({function_name: function_metadata})
return cpp_code
except Exception as e:
logging.error(f"Error generating C++ code for function '{function_name}': {str(e)}")
return None
# utils/generate_file_structure.py
import os
import logging
from config import (DATABASE_PATH, CPP_OUTPUT_DIR, CMAKE_MINIMUM_VERSION, TARGET_NAME, CXX_STANDARD,
ADDITIONAL_LIBRARIES, LLVM_PATH, INCLUDE_DIRECTORIES, DEBUG_MODE, LOG_FILE)
logging.basicConfig(filename=LOG_FILE, level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def generate_files_from_db():
# Connect to database and fetch functions
with sqlite3.connect(DATABASE_PATH) as conn:
cursor = conn.cursor()
project_metadata = gather_project_metadata()
cursor.execute("SELECT function_id, name, cpp_code, offset, signature FROM cpp_generation WHERE validation_status = 'verified'")
functions = cursor.fetchall()
header_content = {}
source_content = {}
for func_id, func_name, cpp_code, offset, signature in functions:
header_file, source_file = determine_file_structure(func_name, project_metadata)
if header_file not in header_content:
header_content[header_file] = ""
header_content[header_file] += f"{signature};\n"
if source_file not in source_content:
source_content[source_file] = ""
source_content[source_file] += cpp_code + "\n"
# Write header and source files
for filename, content in header_content.items():
with open(os.path.join(CPP_OUTPUT_DIR, "include", filename), "w") as f:
f.write("#pragma once\n\n" + content)
for filename, content in source_content.items():
with open(os.path.join(CPP_OUTPUT_DIR, "src", filename), "w") as f:
f.write(content)
generate_cmake_file(project_metadata)
def determine_file_structure(func_name, project_metadata):
module = project_metadata.get("modules", {}).get(func_name)
namespace = project_metadata.get("namespaces", {}).get(func_name)
if module:
module_dir = os.path.join(CPP_OUTPUT_DIR, "src", module)
os.makedirs(module_dir, exist_ok=True)
header_file = os.path.join(module, f"{func_name}.h")
source_file = os.path.join(module, f"{func_name}.cpp")
elif namespace:
namespace_dir = os.path.join(CPP_OUTPUT_DIR, "src", namespace)
os.makedirs(namespace_dir, exist_ok=True)
header_file = os.path.join(namespace, f"{func_name}.h")
source_file = os.path.join(namespace, f"{func_name}.cpp")
else:
header_file = f"{func_name}.h"
source_file = f"{func_name}.cpp"
return header_file, source_file
def generate_cmake_file(project_metadata):
# Build CMake content
library_includes = "\n".join([f"target_link_libraries({TARGET_NAME} {lib})" for lib in ADDITIONAL_LIBRARIES])
include_directories = "\n".join([f"include_directories({dir})" for dir in INCLUDE_DIRECTORIES])
cmake_content = f"""
cmake_minimum_required(VERSION {CMAKE_MINIMUM_VERSION})
project({TARGET_NAME})
set(CMAKE_CXX_STANDARD {CXX_STANDARD})
{include_directories}
# Add sources
file(GLOB SOURCES "src/**/*.cpp")
# Define executable
add_executable({TARGET_NAME} ${{SOURCES}})
# Link libraries
{library_includes}
# LLVM configuration (if needed)
if (EXISTS "{LLVM_PATH}")
find_package(LLVM REQUIRED PATHS "{LLVM_PATH}")
target_include_directories({TARGET_NAME} PRIVATE ${{LLVM_INCLUDE_DIRS}})
target_link_libraries({TARGET_NAME} ${{LLVM_LIBS}})
add_definitions(${{LLVM_DEFINITIONS}})
endif()
"""
with open(os.path.join(CPP_OUTPUT_DIR, "CMakeLists.txt"), "w") as cmake_file:
cmake_file.write(cmake_content)
function_schema.json:
{
"type": "object",
"properties": {
"name": {"type": "string"},
"address": {"type": "string"},
"offset": {"type": "integer"},
"instructions": {
"type": "array",
"items": {"type": "string"}
},
"pseudo_code": {"type": "string"},
"cpp_code": {"type": "string"}
},
"required": ["name", "address", "instructions", "pseudo_code", "cpp_code"]
}
```
Assistant: To incorporate the `similarity_analysis.py` utility into your existing project for further validation techniques using similarity ratios and graph-based comparisons, you will need to adjust the project structure a bit. Here's a general plan on how to do that:
1. **Create `similarity_analysis.py`**:
Implement the logic for calculating similarity ratios and performing graph-based comparisons in a new utility.
```python
# utils/similarity_analysis.py
import logging
from difflib import SequenceMatcher
import networkx as nx # Example: NetworkX for graph-based comparisons
def calculate_similarity(a, b):
"""Calculate similarity ratio between two strings."""
return SequenceMatcher(None, a, b).ratio()
def graph_based_comparison(graph_a, graph_b):
"""Compare two graphs and return a similarity measure."""
# Example: Using NetworkX graph isomorphism, you might need a more
# sophisticated comparison based on your requirement.
matcher = nx.isomorphism.GraphMatcher(graph_a, graph_b)
is_isomorphic = matcher.is_isomorphic()
return is_isomorphic
def run_analysis_func(func_cpp, original_cpp):
"""Run similar analysis on a decompiled function versus original code."""
# Example: Calculate textual similarity
similarity_score = calculate_similarity(func_cpp, original_cpp)
# Example: Compare control flow graph or dependency graph structure
func_graph = nx.DiGraph() # An example graph representation
original_graph = nx.DiGraph() # An example graph representation
graph_similarity = graph_based_comparison(func_graph, original_graph)
return {"text_similarity": similarity_score, "graph_similarity": graph_similarity}
```
2. **Integrate with Main Process**:
Update the necessary functions to call the `run_analysis_func`.
```python
# main.py
from utils.similarity_analysis import run_analysis_func
def main():
gather_project_metadata(BINARY_PATH)
os.makedirs(CPP_OUTPUT_DIR, exist_ok=True)
decompiled_functions = decompile_functions(BINARY_PATH)
for func_name, func_data in decompiled_functions.items():
if validate_function(func_data):
analysis_results = run_analysis_func(func_data['cpp_code'], func_data['original_code']) # Assuming original code is available
if analysis_results['text_similarity'] > 0.8 and analysis_results['graph_similarity']:
save_to_database(func_name, func_data['offset'], func_data['assembly_code'], func_data['decompiled_code'], "")
else:
logging.warning(f"Similarity analysis failed for function '{func_name}'.")
verified_functions = verify_function(BINARY_PATH, decompiled_functions)
feedback_loop(decompiled_functions, verified_functions)
generate_files_from_db()
close_database()
if __name__ == "__main__":
main()
```
3. **Update `generate_cpp_code` to Incorporate Similarity Analysis**:
Ensure the C++ generation function checks similarity, which could now use the enhanced `run_analysis_func` functionality.
```python
# Inside utils/cpp_generator.py
from utils.similarity_analysis import run_analysis_func
def generate_cpp_code(pseudo_c_code, function_name, symbol_names="", comments=""):
try:
# Your existing code
cpp_code = cpp_code_response.get('output', {}).get('cpp_code', '')
# New similarity check using similarity analysis
analysis_results = run_analysis_func(pseudo_c_code, cpp_code)
if not analysis_results['text_similarity'] > 0.9 or not analysis_results['graph_similarity']:
logging.warning(f"Similarity check failed for '{function_name}'. Triggering feedback loop.")
feedback_loop({function_name: function_metadata})
return cpp_code
except Exception as e:
logging.error(f"Error generating C++ code for function '{function_name}': {str(e)}")
return None
```
4. **Incorporate Graph-Based Analysis Where Appropriate**:
Throughout the system, whether validating or verifying, include graph-based techniques to ensure that logic structures (like control flows) match between original and decompiled code.
This outline illustrates how you can start integrating the new `similarity_analysis.py` functionality into your overall decompilation and code generation pipeline. Ensure usage adapts appropriately to real data structures (like constructing graphs based on actual parsing), adjust thresholds for similarity checks based on testing, and feel free to expand the analysis sophistication to suit your project's needs.
User: Let's retry, the metrics are required to check how close we are with C++ generation and the loop feedback for the entire process to automatically bruteforce reverse engineer.
```
Diffing Assembler: Comparing the assembler output of generated C++ against the original binary ensures that the behavior matches at a low level, providing a direct validation check.
Diffing Control Flow Graphs (CFG): Control flow graphs represent function execution paths. Matching these between the decompiled output and the original binary will highlight any missing or extraneous branches, revealing structural inconsistencies.
Similarity Ratio Calculation: Calculating similarity ratios between functions, both at the assembly and C++ levels, can provide an automated metric for assessing whether the LLM-generated code resembles the expected output.
```
```
# main.py
import os
from config import BINARY_PATH, CPP_OUTPUT_DIR
from utils.project_info import gather_project_metadata
from utils.decompilation import decompile_functions
from utils.validation import validate_function
from utils.verification import verify_function
from utils.db_utils import save_to_database, close_database
from utils.generate_file_structure import generate_files_from_db
def main():
gather_project_metadata(BINARY_PATH)
os.makedirs(CPP_OUTPUT_DIR, exist_ok=True)
decompiled_functions = decompile_functions(BINARY_PATH)
for func_name, func_data in decompiled_functions.items():
if validate_function(func_data):
save_to_database(func_name, func_data['offset'], func_data['assembly_code'], func_data['decompiled_code'], "")
verified_functions = verify_function(BINARY_PATH, decompiled_functions)
feedback_loop(decompiled_functions, verified_functions)
generate_files_from_db()
close_database()
if __name__ == "__main__":
main()
# config.py
# Base URL for the Ollama API running on WSL2
OLLAMA_BASE_URL = "http://172.31.72.252:11434"
# Model to use in Ollama
LLM_MODEL = "ollama/qwen2.5:14b"
# Debug mode toggle
DEBUG_MODE = True
# Paths for project files and directories
BINARY_PATH = r"Civ5XP" # Path to the binary file for decompilation
CPP_OUTPUT_DIR = "output" # Directory for C++ output files
LOG_FILE = "logs/debug.log" # Log file path
DATABASE_PATH = "binary_decompilation.db" # Database file path
# Decompilation and C++ generation settings
MAX_CONTEXT_LENGTH = 128000 # Max context length for LLM input
MAX_TOKENS = 8192 # Max tokens for LLM output
CXX_STANDARD = 17 # C++ standard for generated code
# Function call model configuration for LLM
MODEL_FUNCTION_CALL_SETTINGS = {
"name": "generate_cpp",
"parameters": [
{"name": "cpp_code", "type": "string"},
{"name": "warnings", "type": "list"},
{"name": "feedback", "type": "string"}
]
}
# CMake file generation options
CMAKE_MINIMUM_VERSION = "3.10"
TARGET_NAME = "DecompiledProject" # Target name in CMakeLists.txt
ADDITIONAL_LIBRARIES = ["libA", "libB"] # Additional libraries to link in CMakeLists.txt
LLVM_PATH = "/usr/local/llvm" # Adjust to your actual LLVM path if needed
INCLUDE_DIRECTORIES = ["include", "/path/to/other/includes"]
```
```
# utils/decompilation.py
from utils.verification import verify_function, compile_cpp_to_llvm_ir
from utils.validation import validate_function
from utils.feedback_loop import feedback_loop
def decompile_functions(binary_file=BINARY_PATH):
project_metadata = gather_project_metadata(binary_file)
function_summaries = {}
with pyhidra.open_program(binary_file) as flat_api:
program = flat_api.getCurrentProgram()
listing = program.getListing()
decompiler = DecompInterface()
decompiler.openProgram(program)
for function in listing.getFunctions(True):
function_name = function.getName()
function_offset = function.getEntryPoint().getOffset()
db_entry = load_function_from_database(function_name)
if db_entry:
logging.debug(f"Function '{function_name}' already in database.")
continue
try:
results = decompiler.decompileFunction(function, 0, TaskMonitor.DUMMY)
if results and results.decompiledFunction is not None:
decompiled_code = results.getDecompiledFunction().getC()
assembly_code = str(function.getBody())
cpp_code = generate_cpp_code(decompiled_code, function_name)
# Decompile function and collect metadata
function_metadata = {
"name": function_name,
"offset": function_offset,
"assembly_code": assembly_code,
"decompiled_code": decompiled_code,
"symbol_names": [symbol.getName() for symbol in function.getSymbols()],
"comments": extract_comments_from_function(function),
"cpp_code": cpp_code
}
# Store to database after validation
if validate_function(function_metadata):
save_to_database(function_name, function_metadata, 'decompilation_output')
function_summaries[function_name] = function_metadata
# Verification step - LLVM IR and Control Flow Graph matching
llvm_ir = compile_cpp_to_llvm_ir(cpp_code)
verification_result = verify_function(binary_file, {function_name: function_metadata})
if not verification_result:
logging.warning(f"Verification failed for '{function_name}'. Triggering feedback loop.")
feedback_loop({function_name: function_metadata}, function_summaries)
else:
logging.warning(f"Validation failed for '{function_name}'.")
else:
logging.error(f"Failed decompiling '{function_name}'.")
except Exception as e:
logging.error(f"Error processing '{function_name}': {str(e)}")
close_database()
return function_summaries
```
```
# utils/verification.py
import angr
import llvmlite.binding as llvm
import logging
import subprocess
import tempfile
from config import DEBUG_MODE
from utils.db_utils import load_project_metadata
logging.basicConfig(filename="logs/debug.log", level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def verify_function(binary_path, functions_data):
project_metadata = load_project_metadata()
min_address = int(project_metadata.get("min_address", "0"), 16)
max_address = int(project_metadata.get("max_address", "FFFFFFFF"), 16)
for func_name, func_data in functions_data.items():
cpp_code = func_data.get("cpp_code")
if cpp_code:
llvm_ir = compile_cpp_to_llvm_ir(cpp_code)
if llvm_ir:
result = compare_with_binary(binary_path, func_name, llvm_ir, min_address, max_address)
if result:
logging.info(f"Verification passed for '{func_name}'.")
else:
logging.warning(f"Verification failed for '{func_name}'.")
def compile_cpp_to_llvm_ir(cpp_code):
try:
with tempfile.NamedTemporaryFile(suffix=".cpp", delete=False) as cpp_file:
cpp_file.write(cpp_code.encode())
cpp_filename = cpp_file.name
llvm_filename = cpp_filename.replace(".cpp", ".ll")
subprocess.run(["clang", "-emit-llvm", "-S", cpp_filename, "-o", llvm_filename], check=True)
with open(llvm_filename, "r") as llvm_ir_file:
llvm_ir = llvm_ir_file.read()
return llvm_ir
except subprocess.CalledProcessError as e:
logging.error(f"Clang compilation failed: {str(e)}")
return None
except Exception as e:
logging.error(f"Error compiling C++ to LLVM IR: {str(e)}")
return None
def compare_with_binary(binary_path, func_name, llvm_ir, min_address, max_address):
try:
project = angr.Project(binary_path, auto_load_libs=False)
cfg = project.analyses.CFGFast()
llvm_func_addresses = extract_function_addresses_from_llvm(llvm_ir, func_name)
for address in llvm_func_addresses:
if not (min_address <= address <= max_address):
logging.warning(f"Address {address} for '{func_name}' out of valid range.")
continue
binary_func = cfg.kb.functions.get_by_addr(address)
if binary_func is None:
logging.warning(f"Function '{func_name}' at address {address} not found in binary.")
return False
return True
except Exception as e:
logging.error(f"Error comparing LLVM IR with binary: {str(e)}")
return False
def extract_function_addresses_from_llvm(llvm_ir, func_name):
addresses = []
for line in llvm_ir.splitlines():
if func_name in line and "define" in line:
# Use a more accurate regex or parser for proper extraction
# Assuming format like `@func_name = external addrspace(0) constant i32 0x...`
address_match = re.search(r'0x[0-9A-Fa-f]+', line)
if address_match:
address = int(address_match.group(0), 16)
addresses.append(address)
return addresses
```
```
# utils/validation.py
import jsonschema
import logging
import json
from config import DEBUG_MODE
logging.basicConfig(filename="logs/debug.log", level=logging.DEBUG if DEBUG_MODE else logging.INFO)
with open("schemas/function_schema.json") as f:
function_schema = json.load(f)
def validate_function(function_data):
try:
jsonschema.validate(instance=function_data, schema=function_schema)
logging.debug(f"Validation passed: {function_data.get('name', 'unknown')}")
return True
except jsonschema.ValidationError as e:
logging.error(f"Validation error: {e}")
return False
```
```
# utils/db_utils.py
import sqlite3
from pathlib import Path
db_path = Path("binary_decompilation.db")
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Functions table
cursor.execute('''
CREATE TABLE IF NOT EXISTS functions (
function_id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE,
address TEXT,
entry_point TEXT,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Decompilation Output table
cursor.execute('''
CREATE TABLE IF NOT EXISTS decompilation_output (
function_id INTEGER,
assembly_code TEXT,
pseudo_c_code TEXT,
last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (function_id) REFERENCES functions(function_id)
)
''')
# C++ Generation table
cursor.execute('''
CREATE TABLE IF NOT EXISTS cpp_generation (
function_id INTEGER,
cpp_code TEXT,
warnings TEXT,
feedback TEXT,
validation_status TEXT,
last_generated TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (function_id) REFERENCES functions(function_id)
)
''')
# Verification and Feedback table
cursor.execute('''
CREATE TABLE IF NOT EXISTS verification_feedback (
function_id INTEGER,
discrepancies TEXT,
refined_prompt TEXT,
verification_status TEXT,
last_verified TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (function_id) REFERENCES functions(function_id)
)
''')
conn.commit()
# Project Metadata table
cursor.execute('''
CREATE TABLE IF NOT EXISTS project_metadata (
project_id INTEGER PRIMARY KEY AUTOINCREMENT,
project_file_name TEXT,
last_modified TEXT,
readonly BOOLEAN,
program_name TEXT,
language_id TEXT,
compiler_id TEXT,
processor TEXT,
endian TEXT,
address_size INTEGER,
min_address TEXT,
max_address TEXT,
num_bytes INTEGER,
num_memory_blocks INTEGER,
num_instructions INTEGER,
num_defined_data INTEGER,
num_functions INTEGER,
num_symbols INTEGER,
num_data_types INTEGER,
analyzed BOOLEAN,
created_with_ghidra_version TEXT,
file_type TEXT,
file_location TEXT,
elf_original_image_base TEXT,
relocatable BOOLEAN,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Missing Libraries table
cursor.execute('''
CREATE TABLE IF NOT EXISTS missing_libraries (
library_id INTEGER PRIMARY KEY AUTOINCREMENT,
project_id INTEGER,
library_name TEXT,
last_checked TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (project_id) REFERENCES project_metadata(project_id)
)
''')
conn.commit()
def save_project_metadata(metadata):
fields = ', '.join(metadata.keys())
placeholders = ', '.join(['?'] * len(metadata))
values = tuple(metadata.values())
cursor.execute(f'''
INSERT INTO project_metadata ({fields})
VALUES ({placeholders})
''', values)
conn.commit()
def save_missing_library(project_id, library_name):
cursor.execute('''
INSERT INTO missing_libraries (project_id, library_name)
VALUES (?, ?)
''', (project_id, library_name))
conn.commit()
def save_to_database(function_id, data, table):
try:
fields = ', '.join(data.keys())
placeholders = ', '.join(['?'] * len(data))
values = tuple(data.values())
cursor.execute(f'''
INSERT OR REPLACE INTO {table} ({fields})
VALUES ({placeholders})
''', values)
conn.commit()
except sqlite3.Error as e:
print(f"Database error: {e}")
def load_function_from_database(function_name):
cursor.execute('SELECT * FROM functions WHERE name = ?', (function_name,))
return cursor.fetchone()
def close_database():
conn.close()
# context manager for database operations
from contextlib import contextmanager
@contextmanager
def get_db_cursor():
conn = sqlite3.connect(db_path)
try:
yield conn.cursor()
finally:
conn.commit()
conn.close()
```
```
# utils/feedback_loop.py
import logging
from utils.db_utils import save_to_database
from config import DEBUG_MODE
logging.basicConfig(filename="logs/debug.log", level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def log_discrepancies(function_id, discrepancy_details):
logging.debug(f"Logging discrepancy for function_id '{function_id}': {discrepancy_details}")
save_to_database(function_id, {"discrepancies": str(discrepancy_details)}, "verification_feedback")
def refine_prompts(discrepancies):
if not discrepancies:
logging.debug("No discrepancies found; no prompt refinement necessary.")
return
for func_id, details in discrepancies:
logging.debug(f"Refining prompts based on discrepancy for function_id '{func_id}'")
# Adjustments to prompt structure or content based on discrepancy type
if "type_mismatch" in details:
prompt_modifier = "Ensure strict type compliance in generated C++."
elif "missing_code" in details:
prompt_modifier = "Double-check all control flows and branches."
refined_prompt = {
"role": "user",
"content": f"Refine the C++ output based on these observations:\n{details}\n{prompt_modifier}"
}
save_to_database(func_id, {"refined_prompt": str(refined_prompt)}, "verification_feedback")
logging.debug(f"Updated prompt for function_id '{func_id}': {refined_prompt}")
def feedback_loop(decompiled_functions, verified_functions):
discrepancies = []
for func_id, details in decompiled_functions.items():
if func_id not in verified_functions or verified_functions[func_id]["cpp_code"] != details["cpp_code"]:
discrepancy_details = {"decompiled": details, "verified": verified_functions.get(func_id)}
discrepancies.append((func_id, discrepancy_details))
log_discrepancies(func_id, discrepancy_details)
refine_prompts(discrepancies)
```
```
# utils/project_info.py
import os
import logging
import pyhidra
from utils.db_utils import save_project_metadata, save_missing_library
logging.basicConfig(filename="logs/project_info.log", level=logging.INFO)
def gather_project_metadata(binary_path):
with pyhidra.open_program(binary_path) as flat_api:
program = flat_api.getCurrentProgram()
metadata = {
"project_file_name": program.getDomainFile().getName(),
"last_modified": program.getModificationDate().toString(),
"readonly": program.isReadonly(),
"program_name": program.getName(),
"language_id": program.getLanguageID().toString(),
"compiler_id": program.getCompilerSpec().getCompilerSpecID().toString(),
"processor": program.getLanguage().getProcessor().toString(),
"endian": program.getLanguage().isBigEndian(),
"address_size": program.getDefaultPointerSize(),
"min_address": program.getMinAddress().toString(),
"max_address": program.getMaxAddress().toString(),
"num_bytes": program.getMemory().getNumAddresses(),
"num_memory_blocks": program.getMemoryBlockCount(),
"num_instructions": program.getListing().getNumInstructions(),
"num_defined_data": program.getListing().getNumDefinedData(),
"num_functions": program.getFunctionManager().getFunctionCount(),
"num_symbols": program.getSymbolTable().getNumSymbols(),
"num_data_types": len(program.getDataTypeManager().getAllDataTypes()),
"analyzed": program.isAnalyzed(),
"created_with_ghidra_version": program.getVersion(),
"file_type": program.getExecutableFormat(),
"file_location": program.getExecutablePath(),
"elf_original_image_base": program.getExecutableBase(),
"relocatable": program.isRelocatable()
}
save_project_metadata(metadata)
# Log and save required libraries
required_libs = program.getMemory().getExternalLibraries()
for lib in required_libs:
if not os.path.exists(lib):
logging.warning(f"Missing library: {lib}")
save_missing_library(lib)
logging.info("Project metadata and library information gathered and saved.")
```
```
# utils/cpp_generator.py
import logging
from config import LLM_MODEL, DEBUG_MODE, MAX_CONTEXT_LENGTH, MAX_TOKENS, MODEL_FUNCTION_CALL_SETTINGS, LOG_FILE
from litellm import completion
logging.basicConfig(filename=LOG_FILE, level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def generate_cpp_code(pseudo_c_code, function_name, symbol_names="", comments=""):
try:
# Include symbols and comments in the prompt for context-awareness
prompt_content = f"Convert the following pseudo-C code to structured C++:\n{pseudo_c_code}"
if symbol_names or comments:
prompt_content += f"\n\nSymbols:\n{symbol_names}\nComments:\n{comments}"
cpp_code_response = completion(
model=LLM_MODEL,
messages=[{
"role": "user",
"content": prompt_content
}],
format="json",
max_context_length=MAX_CONTEXT_LENGTH,
max_tokens=MAX_TOKENS,
function_call=MODEL_FUNCTION_CALL_SETTINGS
)
# Similarity check and feedback trigger
cpp_code = cpp_code_response.get('output', {}).get('cpp_code', '')
if not check_similarity(pseudo_c_code, cpp_code):
logging.warning(f"Similarity check failed for '{function_name}'. Triggering feedback loop.")
feedback_loop({function_name: function_metadata})
return cpp_code
except Exception as e:
logging.error(f"Error generating C++ code for function '{function_name}': {str(e)}")
return None
```
```
# utils/generate_file_structure.py
import os
import logging
from config import (DATABASE_PATH, CPP_OUTPUT_DIR, CMAKE_MINIMUM_VERSION, TARGET_NAME, CXX_STANDARD,
ADDITIONAL_LIBRARIES, LLVM_PATH, INCLUDE_DIRECTORIES, DEBUG_MODE, LOG_FILE)
logging.basicConfig(filename=LOG_FILE, level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def generate_files_from_db():
# Connect to database and fetch functions
with sqlite3.connect(DATABASE_PATH) as conn:
cursor = conn.cursor()
project_metadata = gather_project_metadata()
cursor.execute("SELECT function_id, name, cpp_code, offset, signature FROM cpp_generation WHERE validation_status = 'verified'")
functions = cursor.fetchall()
header_content = {}
source_content = {}
for func_id, func_name, cpp_code, offset, signature in functions:
header_file, source_file = determine_file_structure(func_name, project_metadata)
if header_file not in header_content:
header_content[header_file] = ""
header_content[header_file] += f"{signature};\n"
if source_file not in source_content:
source_content[source_file] = ""
source_content[source_file] += cpp_code + "\n"
# Write header and source files
for filename, content in header_content.items():
with open(os.path.join(CPP_OUTPUT_DIR, "include", filename), "w") as f:
f.write("#pragma once\n\n" + content)
for filename, content in source_content.items():
with open(os.path.join(CPP_OUTPUT_DIR, "src", filename), "w") as f:
f.write(content)
generate_cmake_file(project_metadata)
def determine_file_structure(func_name, project_metadata):
module = project_metadata.get("modules", {}).get(func_name)
namespace = project_metadata.get("namespaces", {}).get(func_name)
if module:
module_dir = os.path.join(CPP_OUTPUT_DIR, "src", module)
os.makedirs(module_dir, exist_ok=True)
header_file = os.path.join(module, f"{func_name}.h")
source_file = os.path.join(module, f"{func_name}.cpp")
elif namespace:
namespace_dir = os.path.join(CPP_OUTPUT_DIR, "src", namespace)
os.makedirs(namespace_dir, exist_ok=True)
header_file = os.path.join(namespace, f"{func_name}.h")
source_file = os.path.join(namespace, f"{func_name}.cpp")
else:
header_file = f"{func_name}.h"
source_file = f"{func_name}.cpp"
return header_file, source_file
def generate_cmake_file(project_metadata):
# Build CMake content
library_includes = "\n".join([f"target_link_libraries({TARGET_NAME} {lib})" for lib in ADDITIONAL_LIBRARIES])
include_directories = "\n".join([f"include_directories({dir})" for dir in INCLUDE_DIRECTORIES])
cmake_content = f"""
cmake_minimum_required(VERSION {CMAKE_MINIMUM_VERSION})
project({TARGET_NAME})
set(CMAKE_CXX_STANDARD {CXX_STANDARD})
{include_directories}
# Add sources
file(GLOB SOURCES "src/**/*.cpp")
# Define executable
add_executable({TARGET_NAME} ${{SOURCES}})
# Link libraries
{library_includes}
# LLVM configuration (if needed)
if (EXISTS "{LLVM_PATH}")
find_package(LLVM REQUIRED PATHS "{LLVM_PATH}")
target_include_directories({TARGET_NAME} PRIVATE ${{LLVM_INCLUDE_DIRS}})
target_link_libraries({TARGET_NAME} ${{LLVM_LIBS}})
add_definitions(${{LLVM_DEFINITIONS}})
endif()
"""
with open(os.path.join(CPP_OUTPUT_DIR, "CMakeLists.txt"), "w") as cmake_file:
cmake_file.write(cmake_content)
```
```
function_schema.json:
{
"type": "object",
"properties": {
"name": {"type": "string"},
"address": {"type": "string"},
"offset": {"type": "integer"},
"instructions": {
"type": "array",
"items": {"type": "string"}
},
"pseudo_code": {"type": "string"},
"cpp_code": {"type": "string"}
},
"required": ["name", "address", "instructions", "pseudo_code", "cpp_code"]
}
```
```plaintext
# main.py
import os
from config import BINARY_PATH, CPP_OUTPUT_DIR
from utils.project_info import gather_project_metadata
from utils.decompilation import decompile_functions
from utils.validation import validate_function
from utils.verification import verify_function
from utils.db_utils import save_to_database, close_database
from utils.generate_file_structure import generate_files_from_db
def main():
gather_project_metadata(BINARY_PATH)
os.makedirs(CPP_OUTPUT_DIR, exist_ok=True)
decompiled_functions = decompile_functions(BINARY_PATH)
for func_name, func_data in decompiled_functions.items():
if validate_function(func_data):
save_to_database(func_name, func_data['offset'], func_data['assembly_code'], func_data['decompiled_code'], "")
verified_functions = verify_function(BINARY_PATH, decompiled_functions)
feedback_loop(decompiled_functions, verified_functions)
generate_files_from_db()
close_database()
if __name__ == "__main__":
main()
# config.py
# Base URL for the Ollama API running on WSL2
OLLAMA_BASE_URL = "http://172.31.72.252:11434"
# Model to use in Ollama
LLM_MODEL = "ollama/qwen2.5:14b"
# Debug mode toggle
DEBUG_MODE = True
# Paths for project files and directories
BINARY_PATH = r"Civ5XP" # Path to the binary file for decompilation
CPP_OUTPUT_DIR = "output" # Directory for C++ output files
LOG_FILE = "logs/debug.log" # Log file path
DATABASE_PATH = "binary_decompilation.db" # Database file path
# Decompilation and C++ generation settings
MAX_CONTEXT_LENGTH = 128000 # Max context length for LLM input
MAX_TOKENS = 8192 # Max tokens for LLM output
CXX_STANDARD = 17 # C++ standard for generated code
# Function call model configuration for LLM
MODEL_FUNCTION_CALL_SETTINGS = {
"name": "generate_cpp",
"parameters": [
{"name": "cpp_code", "type": "string"},
{"name": "warnings", "type": "list"},
{"name": "feedback", "type": "string"}
]
}
# CMake file generation options
CMAKE_MINIMUM_VERSION = "3.10"
TARGET_NAME = "DecompiledProject" # Target name in CMakeLists.txt
ADDITIONAL_LIBRARIES = ["libA", "libB"] # Additional libraries to link in CMakeLists.txt
LLVM_PATH = "/usr/local/llvm" # Adjust to your actual LLVM path if needed
INCLUDE_DIRECTORIES = ["include", "/path/to/other/includes"]
# utils/decompilation.py
from utils.verification import verify_function, compile_cpp_to_llvm_ir
from utils.validation import validate_function
from utils.feedback_loop import feedback_loop
def decompile_functions(binary_file=BINARY_PATH):
project_metadata = gather_project_metadata(binary_file)
function_summaries = {}
with pyhidra.open_program(binary_file) as flat_api:
program = flat_api.getCurrentProgram()
listing = program.getListing()
decompiler = DecompInterface()
decompiler.openProgram(program)
for function in listing.getFunctions(True):
function_name = function.getName()
function_offset = function.getEntryPoint().getOffset()
db_entry = load_function_from_database(function_name)
if db_entry:
logging.debug(f"Function '{function_name}' already in database.")
continue
try:
results = decompiler.decompileFunction(function, 0, TaskMonitor.DUMMY)
if results and results.decompiledFunction is not None:
decompiled_code = results.getDecompiledFunction().getC()
assembly_code = str(function.getBody())
cpp_code = generate_cpp_code(decompiled_code, function_name)
# Decompile function and collect metadata
function_metadata = {
"name": function_name,
"offset": function_offset,
"assembly_code": assembly_code,
"decompiled_code": decompiled_code,
"symbol_names": [symbol.getName() for symbol in function.getSymbols()],
"comments": extract_comments_from_function(function),
"cpp_code": cpp_code
}
# Store to database after validation
if validate_function(function_metadata):
save_to_database(function_name, function_metadata, 'decompilation_output')
function_summaries[function_name] = function_metadata
# Verification step - LLVM IR and Control Flow Graph matching
llvm_ir = compile_cpp_to_llvm_ir(cpp_code)
verification_result = verify_function(binary_file, {function_name: function_metadata})
if not verification_result:
logging.warning(f"Verification failed for '{function_name}'. Triggering feedback loop.")
feedback_loop({function_name: function_metadata}, function_summaries)
else:
logging.warning(f"Validation failed for '{function_name}'.")
else:
logging.error(f"Failed decompiling '{function_name}'.")
except Exception as e:
logging.error(f"Error processing '{function_name}': {str(e)}")
close_database()
return function_summaries
# utils/verification.py
import angr
import llvmlite.binding as llvm
import logging
import subprocess
import tempfile
from config import DEBUG_MODE
from utils.db_utils import load_project_metadata
logging.basicConfig(filename="logs/debug.log", level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def verify_function(binary_path, functions_data):
project_metadata = load_project_metadata()
min_address = int(project_metadata.get("min_address", "0"), 16)
max_address = int(project_metadata.get("max_address", "FFFFFFFF"), 16)
for func_name, func_data in functions_data.items():
cpp_code = func_data.get("cpp_code")
if cpp_code:
llvm_ir = compile_cpp_to_llvm_ir(cpp_code)
if llvm_ir:
result = compare_with_binary(binary_path, func_name, llvm_ir, min_address, max_address)
if result:
logging.info(f"Verification passed for '{func_name}'.")
else:
logging.warning(f"Verification failed for '{func_name}'.")
def compile_cpp_to_llvm_ir(cpp_code):
try:
with tempfile.NamedTemporaryFile(suffix=".cpp", delete=False) as cpp_file:
cpp_file.write(cpp_code.encode())
cpp_filename = cpp_file.name
llvm_filename = cpp_filename.replace(".cpp", ".ll")
subprocess.run(["clang", "-emit-llvm", "-S", cpp_filename, "-o", llvm_filename], check=True)
with open(llvm_filename, "r") as llvm_ir_file:
llvm_ir = llvm_ir_file.read()
return llvm_ir
except subprocess.CalledProcessError as e:
logging.error(f"Clang compilation failed: {str(e)}")
return None
except Exception as e:
logging.error(f"Error compiling C++ to LLVM IR: {str(e)}")
return None
def compare_with_binary(binary_path, func_name, llvm_ir, min_address, max_address):
try:
project = angr.Project(binary_path, auto_load_libs=False)
cfg = project.analyses.CFGFast()
llvm_func_addresses = extract_function_addresses_from_llvm(llvm_ir, func_name)
for address in llvm_func_addresses:
if not (min_address <= address <= max_address):
logging.warning(f"Address {address} for '{func_name}' out of valid range.")
continue
binary_func = cfg.kb.functions.get_by_addr(address)
if binary_func is None:
logging.warning(f"Function '{func_name}' at address {address} not found in binary.")
return False
return True
except Exception as e:
logging.error(f"Error comparing LLVM IR with binary: {str(e)}")
return False
def extract_function_addresses_from_llvm(llvm_ir, func_name):
addresses = []
for line in llvm_ir.splitlines():
if func_name in line and "define" in line:
# Use a more accurate regex or parser for proper extraction
# Assuming format like `@func_name = external addrspace(0) constant i32 0x...`
address_match = re.search(r'0x[0-9A-Fa-f]+', line)
if address_match:
address = int(address_match.group(0), 16)
addresses.append(address)
return addresses
# utils/validation.py
import jsonschema
import logging
import json
from config import DEBUG_MODE
logging.basicConfig(filename="logs/debug.log", level=logging.DEBUG if DEBUG_MODE else logging.INFO)
with open("schemas/function_schema.json") as f:
function_schema = json.load(f)
def validate_function(function_data):
try:
jsonschema.validate(instance=function_data, schema=function_schema)
logging.debug(f"Validation passed: {function_data.get('name', 'unknown')}")
return True
except jsonschema.ValidationError as e:
logging.error(f"Validation error: {e}")
return False
# utils/db_utils.py
import sqlite3
from pathlib import Path
db_path = Path("binary_decompilation.db")
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Functions table
cursor.execute('''
CREATE TABLE IF NOT EXISTS functions (
function_id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE,
address TEXT,
entry_point TEXT,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Decompilation Output table
cursor.execute('''
CREATE TABLE IF NOT EXISTS decompilation_output (
function_id INTEGER,
assembly_code TEXT,
pseudo_c_code TEXT,
last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (function_id) REFERENCES functions(function_id)
)
''')
# C++ Generation table
cursor.execute('''
CREATE TABLE IF NOT EXISTS cpp_generation (
function_id INTEGER,
cpp_code TEXT,
warnings TEXT,
feedback TEXT,
validation_status TEXT,
last_generated TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (function_id) REFERENCES functions(function_id)
)
''')
# Verification and Feedback table
cursor.execute('''
CREATE TABLE IF NOT EXISTS verification_feedback (
function_id INTEGER,
discrepancies TEXT,
refined_prompt TEXT,
verification_status TEXT,
last_verified TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (function_id) REFERENCES functions(function_id)
)
''')
conn.commit()
# Project Metadata table
cursor.execute('''
CREATE TABLE IF NOT EXISTS project_metadata (
project_id INTEGER PRIMARY KEY AUTOINCREMENT,
project_file_name TEXT,
last_modified TEXT,
readonly BOOLEAN,
program_name TEXT,
language_id TEXT,
compiler_id TEXT,
processor TEXT,
endian TEXT,
address_size INTEGER,
min_address TEXT,
max_address TEXT,
num_bytes INTEGER,
num_memory_blocks INTEGER,
num_instructions INTEGER,
num_defined_data INTEGER,
num_functions INTEGER,
num_symbols INTEGER,
num_data_types INTEGER,
analyzed BOOLEAN,
created_with_ghidra_version TEXT,
file_type TEXT,
file_location TEXT,
elf_original_image_base TEXT,
relocatable BOOLEAN,
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Missing Libraries table
cursor.execute('''
CREATE TABLE IF NOT EXISTS missing_libraries (
library_id INTEGER PRIMARY KEY AUTOINCREMENT,
project_id INTEGER,
library_name TEXT,
last_checked TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (project_id) REFERENCES project_metadata(project_id)
)
''')
conn.commit()
def save_project_metadata(metadata):
fields = ', '.join(metadata.keys())
placeholders = ', '.join(['?'] * len(metadata))
values = tuple(metadata.values())
cursor.execute(f'''
INSERT INTO project_metadata ({fields})
VALUES ({placeholders})
''', values)
conn.commit()
def save_missing_library(project_id, library_name):
cursor.execute('''
INSERT INTO missing_libraries (project_id, library_name)
VALUES (?, ?)
''', (project_id, library_name))
conn.commit()
def save_to_database(function_id, data, table):
try:
fields = ', '.join(data.keys())
placeholders = ', '.join(['?'] * len(data))
values = tuple(data.values())
cursor.execute(f'''
INSERT OR REPLACE INTO {table} ({fields})
VALUES ({placeholders})
''', values)
conn.commit()
except sqlite3.Error as e:
print(f"Database error: {e}")
def load_function_from_database(function_name):
cursor.execute('SELECT * FROM functions WHERE name = ?', (function_name,))
return cursor.fetchone()
def close_database():
conn.close()
# context manager for database operations
from contextlib import contextmanager
@contextmanager
def get_db_cursor():
conn = sqlite3.connect(db_path)
try:
yield conn.cursor()
finally:
conn.commit()
conn.close()
# utils/feedback_loop.py
import logging
from utils.db_utils import save_to_database
from config import DEBUG_MODE
logging.basicConfig(filename="logs/debug.log", level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def log_discrepancies(function_id, discrepancy_details):
logging.debug(f"Logging discrepancy for function_id '{function_id}': {discrepancy_details}")
save_to_database(function_id, {"discrepancies": str(discrepancy_details)}, "verification_feedback")
def refine_prompts(discrepancies):
if not discrepancies:
logging.debug("No discrepancies found; no prompt refinement necessary.")
return
for func_id, details in discrepancies:
logging.debug(f"Refining prompts based on discrepancy for function_id '{func_id}'")
# Adjustments to prompt structure or content based on discrepancy type
if "type_mismatch" in details:
prompt_modifier = "Ensure strict type compliance in generated C++."
elif "missing_code" in details:
prompt_modifier = "Double-check all control flows and branches."
refined_prompt = {
"role": "user",
"content": f"Refine the C++ output based on these observations:\n{details}\n{prompt_modifier}"
}
save_to_database(func_id, {"refined_prompt": str(refined_prompt)}, "verification_feedback")
logging.debug(f"Updated prompt for function_id '{func_id}': {refined_prompt}")
def feedback_loop(decompiled_functions, verified_functions):
discrepancies = []
for func_id, details in decompiled_functions.items():
if func_id not in verified_functions or verified_functions[func_id]["cpp_code"] != details["cpp_code"]:
discrepancy_details = {"decompiled": details, "verified": verified_functions.get(func_id)}
discrepancies.append((func_id, discrepancy_details))
log_discrepancies(func_id, discrepancy_details)
refine_prompts(discrepancies)
# utils/project_info.py
import os
import logging
import pyhidra
from utils.db_utils import save_project_metadata, save_missing_library
logging.basicConfig(filename="logs/project_info.log", level=logging.INFO)
def gather_project_metadata(binary_path):
with pyhidra.open_program(binary_path) as flat_api:
program = flat_api.getCurrentProgram()
metadata = {
"project_file_name": program.getDomainFile().getName(),
"last_modified": program.getModificationDate().toString(),
"readonly": program.isReadonly(),
"program_name": program.getName(),
"language_id": program.getLanguageID().toString(),
"compiler_id": program.getCompilerSpec().getCompilerSpecID().toString(),
"processor": program.getLanguage().getProcessor().toString(),
"endian": program.getLanguage().isBigEndian(),
"address_size": program.getDefaultPointerSize(),
"min_address": program.getMinAddress().toString(),
"max_address": program.getMaxAddress().toString(),
"num_bytes": program.getMemory().getNumAddresses(),
"num_memory_blocks": program.getMemoryBlockCount(),
"num_instructions": program.getListing().getNumInstructions(),
"num_defined_data": program.getListing().getNumDefinedData(),
"num_functions": program.getFunctionManager().getFunctionCount(),
"num_symbols": program.getSymbolTable().getNumSymbols(),
"num_data_types": len(program.getDataTypeManager().getAllDataTypes()),
"analyzed": program.isAnalyzed(),
"created_with_ghidra_version": program.getVersion(),
"file_type": program.getExecutableFormat(),
"file_location": program.getExecutablePath(),
"elf_original_image_base": program.getExecutableBase(),
"relocatable": program.isRelocatable()
}
save_project_metadata(metadata)
# Log and save required libraries
required_libs = program.getMemory().getExternalLibraries()
for lib in required_libs:
if not os.path.exists(lib):
logging.warning(f"Missing library: {lib}")
save_missing_library(lib)
logging.info("Project metadata and library information gathered and saved.")
# utils/cpp_generator.py
import logging
from config import LLM_MODEL, DEBUG_MODE, MAX_CONTEXT_LENGTH, MAX_TOKENS, MODEL_FUNCTION_CALL_SETTINGS, LOG_FILE
from litellm import completion
logging.basicConfig(filename=LOG_FILE, level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def generate_cpp_code(pseudo_c_code, function_name, symbol_names="", comments=""):
try:
# Include symbols and comments in the prompt for context-awareness
prompt_content = f"Convert the following pseudo-C code to structured C++:\n{pseudo_c_code}"
if symbol_names or comments:
prompt_content += f"\n\nSymbols:\n{symbol_names}\nComments:\n{comments}"
cpp_code_response = completion(
model=LLM_MODEL,
messages=[{
"role": "user",
"content": prompt_content
}],
format="json",
max_context_length=MAX_CONTEXT_LENGTH,
max_tokens=MAX_TOKENS,
function_call=MODEL_FUNCTION_CALL_SETTINGS
)
# Similarity check and feedback trigger
cpp_code = cpp_code_response.get('output', {}).get('cpp_code', '')
if not check_similarity(pseudo_c_code, cpp_code):
logging.warning(f"Similarity check failed for '{function_name}'. Triggering feedback loop.")
feedback_loop({function_name: function_metadata})
return cpp_code
except Exception as e:
logging.error(f"Error generating C++ code for function '{function_name}': {str(e)}")
return None
# utils/generate_file_structure.py
import os
import logging
from config import (DATABASE_PATH, CPP_OUTPUT_DIR, CMAKE_MINIMUM_VERSION, TARGET_NAME, CXX_STANDARD,
ADDITIONAL_LIBRARIES, LLVM_PATH, INCLUDE_DIRECTORIES, DEBUG_MODE, LOG_FILE)
logging.basicConfig(filename=LOG_FILE, level=logging.DEBUG if DEBUG_MODE else logging.INFO)
def generate_files_from_db():
# Connect to database and fetch functions
with sqlite3.connect(DATABASE_PATH) as conn:
cursor = conn.cursor()
project_metadata = gather_project_metadata()
cursor.execute("SELECT function_id, name, cpp_code, offset, signature FROM cpp_generation WHERE validation_status = 'verified'")
functions = cursor.fetchall()
header_content = {}
source_content = {}
for func_id, func_name, cpp_code, offset, signature in functions:
header_file, source_file = determine_file_structure(func_name, project_metadata)
if header_file not in header_content:
header_content[header_file] = ""
header_content[header_file] += f"{signature};\n"
if source_file not in source_content:
source_content[source_file] = ""
source_content[source_file] += cpp_code + "\n"
# Write header and source files
for filename, content in header_content.items():
with open(os.path.join(CPP_OUTPUT_DIR, "include", filename), "w") as f:
f.write("#pragma once\n\n" + content)
for filename, content in source_content.items():
with open(os.path.join(CPP_OUTPUT_DIR, "src", filename), "w") as f:
f.write(content)
generate_cmake_file(project_metadata)
def determine_file_structure(func_name, project_metadata):
module = project_metadata.get("modules", {}).get(func_name)
namespace = project_metadata.get("namespaces", {}).get(func_name)
if module:
module_dir = os.path.join(CPP_OUTPUT_DIR, "src", module)
os.makedirs(module_dir, exist_ok=True)
header_file = os.path.join(module, f"{func_name}.h")
source_file = os.path.join(module, f"{func_name}.cpp")
elif namespace:
namespace_dir = os.path.join(CPP_OUTPUT_DIR, "src", namespace)
os.makedirs(namespace_dir, exist_ok=True)
header_file = os.path.join(namespace, f"{func_name}.h")
source_file = os.path.join(namespace, f"{func_name}.cpp")
else:
header_file = f"{func_name}.h"
source_file = f"{func_name}.cpp"
return header_file, source_file
def generate_cmake_file(project_metadata):
# Build CMake content
library_includes = "\n".join([f"target_link_libraries({TARGET_NAME} {lib})" for lib in ADDITIONAL_LIBRARIES])
include_directories = "\n".join([f"include_directories({dir})" for dir in INCLUDE_DIRECTORIES])
cmake_content = f"""
cmake_minimum_required(VERSION {CMAKE_MINIMUM_VERSION})
project({TARGET_NAME})
set(CMAKE_CXX_STANDARD {CXX_STANDARD})
{include_directories}
# Add sources
file(GLOB SOURCES "src/**/*.cpp")
# Define executable
add_executable({TARGET_NAME} ${{SOURCES}})
# Link libraries
{library_includes}
# LLVM configuration (if needed)
if (EXISTS "{LLVM_PATH}")
find_package(LLVM REQUIRED PATHS "{LLVM_PATH}")
target_include_directories({TARGET_NAME} PRIVATE ${{LLVM_INCLUDE_DIRS}})
target_link_libraries({TARGET_NAME} ${{LLVM_LIBS}})
add_definitions(${{LLVM_DEFINITIONS}})
endif()
"""
with open(os.path.join(CPP_OUTPUT_DIR, "CMakeLists.txt"), "w") as cmake_file:
cmake_file.write(cmake_content)
function_schema.json:
{
"type": "object",
"properties": {
"name": {"type": "string"},
"address": {"type": "string"},
"offset": {"type": "integer"},
"instructions": {
"type": "array",
"items": {"type": "string"}
},
"pseudo_code": {"type": "string"},
"cpp_code": {"type": "string"}
},
"required": ["name", "address", "instructions", "pseudo_code", "cpp_code"]
}
```
Assistant: