#!/usr/bin/env python3 """Ghidra headless analysis script for automated binary analysis. Runs Ghidra's analyzeHeadless to perform automated analysis on a binary, extracting functions, strings, imports, exports, and cross-references. """ from __future__ import annotations import argparse import json import os import subprocess import sys import tempfile import re from pathlib import Path from datetime import datetime def find_ghidra_headless(ghidra_path: str) -> str: """Locate the analyzeHeadless script within a Ghidra installation.""" candidates = [ os.path.join(ghidra_path, "support", "analyzeHeadless"), os.path.join(ghidra_path, "support", "analyzeHeadless.bat"), os.path.join(ghidra_path, "analyzeHeadless"), ] for candidate in candidates: if os.path.isfile(candidate): return candidate raise FileNotFoundError( f"Could not find analyzeHeadless in {ghidra_path}. " "Ensure Ghidra is installed and the path is correct." ) def create_ghidra_script(script_dir: str) -> str: """Create a Ghidra post-analysis script that exports analysis data.""" script_content = '''// Ghidra post-analysis script to export analysis data // @category MalwareAnalysis import ghidra.app.script.GhidraScript; import ghidra.program.model.listing.*; import ghidra.program.model.symbol.*; import java.io.*; import com.google.gson.*; public class ExportAnalysis extends GhidraScript { @Override public void run() throws Exception { JsonObject result = new JsonObject(); // Export functions JsonArray functions = new JsonArray(); FunctionIterator funcIter = currentProgram.getFunctionManager().getFunctions(true); while (funcIter.hasNext()) { Function func = funcIter.next(); JsonObject funcObj = new JsonObject(); funcObj.addProperty("name", func.getName()); funcObj.addProperty("address", func.getEntryPoint().toString()); funcObj.addProperty("size", func.getBody().getNumAddresses()); functions.add(funcObj); } result.add("functions", functions); // Export strings JsonArray strings = new JsonArray(); DataIterator dataIter = currentProgram.getListing().getDefinedData(true); while (dataIter.hasNext()) { Data data = dataIter.next(); if (data.getDataType().getName().toLowerCase().contains("string")) { JsonObject strObj = new JsonObject(); strObj.addProperty("address", data.getAddress().toString()); strObj.addProperty("value", data.getDefaultValueRepresentation()); strings.add(strObj); } } result.add("strings", strings); String outputPath = System.getProperty("analysis.output", "/tmp/ghidra_output.json"); try (FileWriter fw = new FileWriter(outputPath)) { fw.write(new Gson().toJson(result)); } println("Analysis exported to: " + outputPath); } } ''' script_path = os.path.join(script_dir, "ExportAnalysis.java") with open(script_path, "w") as f: f.write(script_content) return script_path def run_ghidra_analysis(binary_path: str, ghidra_path: str, output_path: str, timeout: int = 600) -> dict: """Run Ghidra headless analysis on the specified binary.""" headless = find_ghidra_headless(ghidra_path) with tempfile.TemporaryDirectory(prefix="ghidra_project_") as project_dir: script_dir = tempfile.mkdtemp(prefix="ghidra_scripts_") create_ghidra_script(script_dir) output_file = os.path.join(project_dir, "analysis_output.json") cmd = [ headless, project_dir, "MalwareAnalysis", "-import", binary_path, "-postScript", "ExportAnalysis.java", "-scriptPath", script_dir, "-deleteProject", "-analysisTimeoutPerFile", str(timeout), f"-Danalysis.output={output_file}", ] print(f"[*] Running Ghidra headless analysis on: {binary_path}") print(f"[*] Command: {' '.join(cmd)}") try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=timeout + 60, ) analysis_result = { "binary": os.path.basename(binary_path), "binary_path": os.path.abspath(binary_path), "timestamp": datetime.utcnow().isoformat(), "ghidra_returncode": result.returncode, } if os.path.isfile(output_file): with open(output_file, "r") as f: exported_data = json.load(f) analysis_result.update(exported_data) else: analysis_result["warning"] = "Export script did not produce output" analysis_result["stdout"] = result.stdout[-2000:] if result.stdout else "" analysis_result["stderr"] = result.stderr[-2000:] if result.stderr else "" return analysis_result except subprocess.TimeoutExpired: return { "binary": os.path.basename(binary_path), "error": f"Analysis timed out after {timeout} seconds", } except FileNotFoundError: return { "binary": os.path.basename(binary_path), "error": "Ghidra analyzeHeadless not found. Check installation path.", } def parse_ghidra_log(log_text: str) -> dict: """Parse Ghidra log output for summary information.""" summary = { "functions_found": 0, "imports_resolved": 0, "warnings": [], "errors": [], } for line in log_text.split("\n"): if "Number of functions" in line: match = re.search(r"(\d+)", line) if match: summary["functions_found"] = int(match.group(1)) elif "WARNING" in line: summary["warnings"].append(line.strip()) elif "ERROR" in line: summary["errors"].append(line.strip()) return summary def main() -> None: parser = argparse.ArgumentParser( description="Run Ghidra headless analysis on a binary file" ) parser.add_argument( "--input", "--binary", "-b", dest="binary", required=True, help="Path to the binary file to analyze", ) parser.add_argument( "--ghidra-path", "-g", default=os.environ.get("GHIDRA_HOME", "/opt/ghidra"), help="Path to Ghidra installation directory (default: /opt/ghidra or $GHIDRA_HOME)", ) parser.add_argument( "--output", "-o", default="ghidra_analysis.json", help="Output JSON file path (default: ghidra_analysis.json)", ) parser.add_argument( "--timeout", "-t", type=int, default=600, help="Analysis timeout in seconds (default: 600)", ) parser.add_argument( "--format", default="json", choices=["json", "text", "csv"], help="Output format (default: json)", ) args = parser.parse_args() if not os.path.isfile(args.binary): print(f"[!] Error: Binary file not found: {args.binary}", file=sys.stderr) sys.exit(1) try: result = run_ghidra_analysis( binary_path=args.binary, ghidra_path=args.ghidra_path, output_path=args.output, timeout=args.timeout, ) with open(args.output, "w") as f: json.dump(result, f, indent=2) print(f"\n[+] Analysis complete. Results saved to: {args.output}") if "functions" in result: print(f"[+] Functions discovered: {len(result['functions'])}") if "strings" in result: print(f"[+] Strings extracted: {len(result['strings'])}") if "error" in result: print(f"[!] Error: {result['error']}", file=sys.stderr) sys.exit(1) except Exception as e: print(f"[!] Fatal error: {e}", file=sys.stderr) sys.exit(1) if __name__ == "__main__": main()