Kaikuo Xu

from hashlib import sha256
import inspect
from pathlib import Path
import subprocess
import json
import subprocess
import json
import os
import tempfile
import uuid
import shutil
import time

from dotenv import load_dotenv

from redb.extractors.enum import Tag
from redb.models.dataclasses import Decompiled
from redb.extractors.extractor import Extractor


class DecompileGhidra(Extractor):
    def __init__(
        self,
        filepath,
        log,
        index_prefix=None,
        elastic_index=None,
        known_benign=False,
        known_malicious=False,
    ):
        super().__init__(
            filepath, log, index_prefix, elastic_index, known_benign, known_malicious
        )
        self.log.debug(inspect.currentframe().f_code.co_name)
        self.elastic_index = self.index_prefix + "-ghidra"
        self.ghidra_path = "/opt/ghidra"
        self.java_script_path = (
            self.ghidra_path
            + "/Ghidra/Features/Base/ghidra_scripts/GhidraDecompilerScript.java"
        )
        self.decompiled = None
        load_dotenv()
        self.decompiled_folder = os.getenv("DECOMPILED_FOLDER", "/opt/decompiled")
        self.log.debug(f"Decompiled folder: {self.decompiled_folder}")

    def run_command(self, cmd, env=None):
        try:
            self.log.info(f"Starting command: {' '.join(cmd)}")
            start_time = time.time()
            TIMEOUT = 1200  # 20 minutes in seconds
            process = subprocess.Popen(
                cmd, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
            )

            while True:
                output = process.stdout.readline()
                if output:
                    print(output.strip())
                if process.poll() is not None:
                    break
            try:
                stdout, stderr = process.communicate(timeout=TIMEOUT)
            except subprocess.TimeoutExpired:
                process.kill()
                self.log.error(f"Ghidra timed out after {TIMEOUT} seconds")
                # raise subprocess.TimeoutExpired(process.args, TIMEOUT)
                return None
            end_time = time.time()

            self.log.debug(
                f"Command finished. Execution time: {end_time - start_time:.2f} seconds"
            )
            self.log.debug(f"Return code: {process.returncode}")

            if process.returncode != 0:
                self.log.error(f"Error output:\n{stderr}")
                return None
            return stdout
        except Exception as e:
            self.log.error(f"Error running command {' '.join(cmd)}: {e}")
            return None

    def analyze_binary(self):
        self.log.debug(f"Ghidra path: {self.ghidra_path}")
        self.log.debug(f"Binary path: {self.filepath}")
        self.log.debug(f"Java script path: {self.java_script_path}")

        # Check if Java script exists
        if not os.path.exists(self.java_script_path):
            self.log.error(f"Error: Java script not found at {self.java_script_path}")
            return None

        # Set up environment variables
        env = os.environ.copy()
        java_home = "/usr/lib/jvm/java-17-openjdk-amd64"  # Adjust this path if needed
        env["JAVA_HOME"] = java_home
        env["PATH"] = f"{java_home}/bin:{env['PATH']}"
        env["LD_LIBRARY_PATH"] = f"{java_home}/lib:{env.get('LD_LIBRARY_PATH', '')}"
        env["DECOMPILED_FOLDER"] = self.decompiled_folder

        # Print environment variables for debugging
        self.log.debug(f"JAVA_HOME: {env['JAVA_HOME']}")
        self.log.debug(f"PATH: {env['PATH']}")
        self.log.debug(f"LD_LIBRARY_PATH: {env['LD_LIBRARY_PATH']}")

        # Check Ghidra installation
        analyzeHeadless_path = f"{self.ghidra_path}/support/analyzeHeadless"
        self.log.debug(
            f"analyzeHeadless exists: {os.path.exists(analyzeHeadless_path)}"
        )

        # Create a temporary project directory
        project_path = tempfile.gettempdir() + "/ghidra_" + str(uuid.uuid4())
        os.makedirs(project_path, exist_ok=True)
        self.log.debug(f"Created temporary project path: {project_path}")
        output_file = ""

        try:
            # Run Ghidra's headless analyzer
            analyze_cmd = [
                analyzeHeadless_path,
                project_path,
                "TempProject",
                "-import",
                self.filepath,
                "-postScript",
                self.java_script_path,
                self.sha256,
                "-deleteProject",
            ]

            result = self.run_command(analyze_cmd, env=env)
            if result is None:
                return None

            # Read the output JSON file
            output_file = os.path.join(
                self.decompiled_folder, self.sha256 + "-decompiled.json"
            )
            if os.path.exists(output_file):
                with open(output_file, "r") as f:
                    functions = json.load(f)
                return functions
            else:
                self.log.error(
                    f"Output file {output_file} not found. Ghidra analysis may have failed."
                )
                return None
        finally:
            # Clean up
            if os.path.exists(project_path):
                shutil.rmtree(project_path)
                self.log.debug(f"Deleted temporary project path: {project_path}")

    def extract(self):
        self.log.debug(inspect.currentframe().f_code.co_name)
        try:
            functions = self.analyze_binary()

            if functions:
                self.log.info(f"Extracted functions from {self.filepath}:")
                for func in functions:
                    id = sha256(func["address"].encode()).hexdigest()
                    self.decompiled = Decompiled(
                        _id=id,
                        decompiled_function_name=func["name"],
                        decompiled_function_address=func["address"],
                        decompiled_function=func["decompiled"],
                    )
                    self.export_to_elastic([self.decompiled])
            else:
                self.log.error("No decompiled functions extracted.")
            return True
        except Exception as e:
            self.log.error(f"Error extracting decompiled information: {e}")
            return None

    def tag(self):
        return Tag.DECOMPILED.value