Habib Ammari

from hashlib import sha256, md5
import inspect
import subprocess
import json
import os
import time
from datetime import datetime, timezone
from typing import Dict, List, Any, Optional

from dotenv import load_dotenv
from redb.extractors.enum import Tag
from redb.extractors.extractor import Extractor
import magic
import pefile
import ppdeep
import tlsh


class DecompileGhidra(Extractor):
    def __init__(
        self,
        filepath,
        log,
        exporters=None,
        index_prefix=None,
        elastic_index=None,
        known_benign=False,
        known_malicious=False,
        filetype=None,
    ):
        super().__init__(
            filepath,
            log,
            exporters,
            index_prefix,
            elastic_index,
            known_benign,
            known_malicious,
        )
        self.log.debug(inspect.currentframe().f_code.co_name)
        self.ghidra_path = os.getenv("GHIDRA_PATH", "/opt/ghidra")
        self.java_script_path = os.getenv(
            "GHIDRA_SCRIPT_PATH",
            "/opt/ghidra/Ghidra/Features/Base/ghidra_scripts/GhidraDecompilerScript.java",
        )
        self.analysis_results = None
        self.ghidra_process = None  # Track the current process
        self.project_path = None
        self.filetype = filetype

        # Convert TIMEOUT to integer with a default of 1200 seconds (20 minutes)
        try:
            self.TIMEOUT = int(os.getenv("GHIDRA_TIMEOUT", "1200"))
        except ValueError:
            self.log.warning(
                "Invalid GHIDRA_TIMEOUT value, using default of 1200 seconds"
            )
            self.TIMEOUT = 1200

        self.initialize_project()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.cleanup_run()

    def is_dotnet(self):
        try:
            if self.filetype == "pebin":
                file_type = magic.from_buffer(self.binary)
                if ".Net" in file_type:
                    return True
                pe = pefile.PE(self.filepath)
                for entry in pe.OPTIONAL_HEADER.DATA_DIRECTORY:
                    # IMAGE_DIRECTORY_ENTRY_COM_DESCRIPTOR is typically 14
                    if (
                        entry.name == "IMAGE_DIRECTORY_ENTRY_COM_DESCRIPTOR"
                        and entry.Size > 0
                    ):
                        return True
                return False
        except AttributeError as e:
            self.log.error(
                f"AttributeError error dotnet file {self.hash.sha256} Full error : {e}"
            )
            return False

    def cleanup_run(self):
        """Clean up after analysis."""
        try:
            if self.ghidra_process and self.ghidra_process.poll() is None:
                self.ghidra_process.terminate()
                try:
                    self.ghidra_process.wait(timeout=5)
                except subprocess.TimeoutExpired:
                    self.ghidra_process.kill()

            # Clean up project directory
            if self.project_path and os.path.exists(self.project_path):
                import shutil

                shutil.rmtree(self.project_path)
                self.log.debug(f"Cleaned up project directory: {self.project_path}")

            # Force garbage collection
            import gc

            gc.collect()
        except Exception as e:
            self.log.error(f"Error in cleanup: {e}")

    # @classmethod
    # def cleanup_batch(cls):
    #     """Clean up the persistent project at the end of a batch."""
    #     print(f"Cleaning up Ghidra project for batch")
    #     if cls._project_path and os.path.exists(cls._project_path):
    #         try:
    #             import shutil
    #             shutil.rmtree(cls._project_path)
    #             cls._project_initialized = False
    #             cls._project_path = None
    #         except Exception as e:
    #             print(f"Error cleaning up project: {e}")

    def _get_environment(self):
        """Setup and return the environment for Ghidra."""
        env = os.environ.copy()
        java_home = os.getenv("GHIDRA_JAVA_HOME", "/usr/lib/jvm/java-17-openjdk-amd64")
        env.update(
            {
                "JAVA_HOME": java_home,
                "PATH": f"{java_home}/bin:{env['PATH']}",
                "LD_LIBRARY_PATH": f"{java_home}/lib:{env.get('LD_LIBRARY_PATH', '')}",
            }
        )
        # 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']}")

        return env

    def initialize_project(self):
        """Initialize a temporary Ghidra project for this file."""
        # Create unique project directory
        self.project_path = f"/tmp/ghidra_{os.path.basename(self.filepath)}_{str(int(time.time()))}_{os.getpid()}"
        os.makedirs(self.project_path, exist_ok=True)
        self.log.debug(f"Created temporary project at {self.project_path}")

        # Create a minimal initialization file
        init_file = os.path.join(self.project_path, ".init")
        with open(init_file, "wb") as f:
            f.write(bytes([0x7F, 0x45, 0x4C, 0x46]))  # Valid ELF header magic bytes

        # Initialize project with minimal file
        env = self._get_environment()
        cmd = [
            f"{self.ghidra_path}/support/analyzeHeadless",
            self.project_path,
            "TempProject",
            "-import",
            init_file,
        ]

        try:
            result = subprocess.run(cmd, env=env, capture_output=True, text=True)
            if result.returncode != 0:
                self.log.error(f"Failed to initialize project: {result.stderr}")
                raise RuntimeError("Project initialization failed")

            # Clean up initialization file
            os.remove(init_file)
            self.log.debug("Project initialized successfully")

        except Exception as e:
            self.log.error(f"Error initializing project: {e}")
            raise

    def analyze_binary(self) -> Optional[Dict[str, Any]]:
        """Run Ghidra analysis and return results."""
        self.log.debug("Starting binary analysis")

        # # Check if packed
        # if self.check_binary_protection():
        #     self.log.warning("Skipping protected binary")
        #     return None

        # Check for .NET only if needed
        # if self.is_dotnet():
        #     self.MAX_NAMED_ARG_WARNINGS = 10000  # Higher threshold for .NET
        #     self.log.info("Adjusting parameters for .NET binary")
        # else:
        #     self.MAX_NAMED_ARG_WARNINGS = 1000  # Normal threshold

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

        env = self._get_environment()

        try:
            base_cmd = [
                f"{self.ghidra_path}/support/analyzeHeadless",
                self.project_path,
                "TempProject",
                "-import",
                self.filepath,
                "-scriptPath",
                os.path.dirname(self.java_script_path),
                "-postScript",
                self.java_script_path,
                self.sha256,
                self.filepath,
            ]
            return self.run_ghidra(base_cmd, env)

        except Exception as e:
            self.log.error(f"Error in Ghidra analysis: {e}")
            return None

        finally:
            self.cleanup_run()

    def run_ghidra(
        self, cmd: list, env: Optional[Dict[str, str]] = None
    ) -> Optional[Dict[str, Any]]:
        """Run Ghidra process and capture JSON output with improved logging separation."""
        process = None
        try:
            self.log.info(f"Starting Ghidra analysis: {' '.join(cmd)}")
            start_time = time.time()

            process = subprocess.Popen(
                cmd, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
            )
            self.ghidra_process = process

            warning_counter = 0
            named_arg_counter = 0
            # Read all output lines
            json_output = None
            while True:
                line = process.stdout.readline()
                if not line and process.poll() is not None:
                    break

                stripped_line = line.strip()
                if not stripped_line:
                    continue

                # if 'Invalid FieldOrProp value in NamedArg' in stripped_line:
                #     named_arg_counter += 1
                #     if named_arg_counter > self.MAX_NAMED_ARG_WARNINGS:
                #         self.log.error(f"Too many NamedArg warnings ({named_arg_counter}), possible protected file.")
                #         self.ghidra_process.kill()
                #         return None
                if (
                    stripped_line.startswith("{")
                    and '"sha256"' in stripped_line
                    and '"decompiled"' in stripped_line
                ):
                    # This is our actual JSON output from GhidraDecompilerScript
                    json_output = stripped_line
                elif any(level in stripped_line for level in ["INFO", "WARN", "ERROR"]):
                    # Ghidra framework logging
                    log_level = (
                        "debug"
                        if "INFO" in stripped_line
                        else "warning"
                        if "WARN" in stripped_line
                        else "error"
                    )
                    if log_level == "warning" and any(
                        expected in stripped_line
                        for expected in [
                            "Unable to disassemble EXTERNAL block",
                            "Failed to markup ELF Note",
                            "Invalid FieldOrProp value in NamedArg",
                            "Unable to resolve constructor",
                            "Could not follow disassembly flow into non-existing memory",
                            "Unable to read bytes at ram",
                        ]
                    ):
                        # Skip expected warnings
                        continue

                    getattr(self.log, log_level)(f"Ghidra info: {stripped_line}")

            # Process completion and stderr
            try:
                stderr = process.stderr.read()
                process.wait(timeout=self.TIMEOUT)

                if stderr:
                    for line in stderr.splitlines():
                        stripped_line = line.strip()
                        if not stripped_line:
                            continue
                        if "ERROR" in stripped_line:
                            self.log.error(f"Ghidra stderr: {stripped_line}")
                        elif "WARN" in stripped_line:
                            self.log.warning(f"Ghidra stderr: {stripped_line}")
                        else:
                            self.log.debug(f"Ghidra stderr: {stripped_line}")

            except subprocess.TimeoutExpired:
                process.kill()
                self.log.error("Ghidra analysis timed out")
                return None

            elapsed_time = time.time() - start_time
            self.log.debug(f"Ghidra analysis completed in {elapsed_time:.2f}s")

            # Parse JSON output if we found it
            if json_output:
                try:
                    result = json.loads(json_output)
                    # Validate the required structure
                    if not isinstance(result, dict) or not all(
                        k in result
                        for k in ["sha256", "decompiled", "disassembled", "cfg"]
                    ):
                        self.log.error("Invalid JSON structure from Ghidra")
                        return None
                    return result
                except json.JSONDecodeError as e:
                    self.log.error(f"Failed to parse Ghidra JSON output: {e}")
                    return None
            else:
                self.log.error("No JSON output received from Ghidra")
                return None

        except Exception as e:
            self.log.error(f"Error running Ghidra: {str(e)}")
            if hasattr(e, "__traceback__"):
                import traceback

                self.log.debug(
                    f"Traceback: {''.join(traceback.format_tb(e.__traceback__))}"
                )
            return None

        finally:
            if process:
                try:
                    # Ensure pipes are closed
                    if process.stdout:
                        process.stdout.close()
                    if process.stderr:
                        process.stderr.close()
                    # Terminate process if still running
                    if process.poll() is None:
                        process.terminate()
                        try:
                            process.wait(timeout=5)
                        except subprocess.TimeoutExpired:
                            process.kill()
                except Exception as e:
                    self.log.error(f"Error cleaning up Ghidra process: {e}")

    def extract(self) -> bool:
        """Extract and process all analysis results."""
        self.log.debug(inspect.currentframe().f_code.co_name)
        try:
            results = self.analyze_binary()
            if not results:
                return False

            self.analysis_results = results
            return True

        except Exception as e:
            self.log.error(f"Error in extraction: {e}")
            return False

    def prepare_export_data(self, exporter_type: str) -> Any:
        """Prepare data for database export."""
        self.log.debug(inspect.currentframe().f_code.co_name)
        if not self.analysis_results:
            return None

        if exporter_type == "ClickHouseExporter":
            now = datetime.now(timezone.utc)

            def prepare_array_field(value, array_type):
                """Helper to prepare array fields with proper null handling"""
                if value is None:
                    return []
                return value

            def ssdeep_disassembly(func):
                try:
                    if len(func) > 1:
                        return ppdeep.hash(func)
                    return ""
                except Exception as e:
                    self.log.error(f"Error in disassembly ssdeep hash calculation: {e}")
                    return ""

            def tlsh_disassembly(func):
                try:
                    if len(func) >= 50:
                        return tlsh.hash(func.encode("utf-8"))
                    return ""
                except Exception as e:
                    self.log.error(f"Error in disassembly tlsh hash calculation: {e}")
                    return ""

            return {
                "multi_table": True,
                "decompiled_content": {
                    "table": "decompiled_functions_content",
                    "data": [
                        [
                            f["decompiled_content_hash"],
                            f["decompiled_function"],
                            f["function_type"],
                            now,
                        ]
                        for f in self.analysis_results["decompiled"]
                    ],
                    "column_names": [
                        "decompiled_content_hash",
                        "decompiled_function",
                        "function_type",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "String",
                        "Enum8('USER'=1, 'LIBRARY'=2, 'THUNK'=3, 'EXTERNAL'=4, 'UNKNOWN'=5)",
                        "DateTime64(3, 'UTC')",
                    ],
                },
                "decompiled_refs": {
                    "table": "decompiled_functions_references",
                    "data": [
                        [
                            self.analysis_results["sha256"],
                            f["decompiled_content_hash"],
                            f["decompiled_function_name"],
                            f["decompiled_function_address"],
                            now,
                        ]
                        for f in self.analysis_results["decompiled"]
                    ],
                    "column_names": [
                        "sha256",
                        "decompiled_content_hash",
                        "decompiled_function_name",
                        "decompiled_function_address",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "FixedString(64)",
                        "LowCardinality(String)",
                        "String",
                        "DateTime64(3, 'UTC')",
                    ],
                },
                "disassembled_content": {
                    "table": "disassembled_functions_content",
                    "data": [
                        [
                            f["disassembled_content_hash"],
                            f["fully_normalized_content_hash"],
                            f["api_normalized_content_hash"],
                            f["category_normalized_content_hash"],
                            f.get("disassembled_function", ""),
                            f.get("fully_normalized_disassembly", ""),
                            f.get("api_normalized_disassembly", ""),
                            f.get("category_normalized_disassembly", ""),
                            ssdeep_disassembly(f.get("disassembled_function", "")),
                            tlsh_disassembly(f.get("disassembled_function", "")),
                            ssdeep_disassembly(
                                f.get("fully_normalized_disassembly", "")
                            ),
                            tlsh_disassembly(f.get("fully_normalized_disassembly", "")),
                            f.get("function_type", "UNKNOWN"),
                            f.get("instruction_count", 0),
                            prepare_array_field(
                                f.get("instruction_types"), "LowCardinality(String)"
                            ),
                            f.get("control_flow_count", 0),
                            prepare_array_field(
                                f.get("memory_access_pattern"), "LowCardinality(String)"
                            ),
                            prepare_array_field(
                                f.get("register_usage"), "LowCardinality(String)"
                            ),
                            f.get("data_references_count", 0),
                            # prepare_array_field(f.get('opcode_frequency_vector'), 'Float32'),
                            # prepare_array_field(f.get('api_calls_vector'), 'Float32'),
                            # prepare_array_field(f.get('minhash_signature'), 'UInt64'),
                            # f.get('pic_hash', ''),
                            f.get("max_block_size", 0),
                            f.get("num_calls", 0),
                            f.get("stack_size", 0),
                            # prepare_array_field(f.get('instruction_type_ratios'), 'Float32'),
                            # prepare_array_field(f.get('instruction_embedding'), 'Float32'),
                            now,
                        ]
                        for f in self.analysis_results["disassembled"]
                    ],
                    "column_names": [
                        "disassembled_content_hash",
                        "fully_normalized_content_hash",
                        "api_normalized_content_hash",
                        "category_normalized_content_hash",
                        "disassembled_function",
                        "fully_normalized_disassembly",
                        "api_normalized_disassembly",
                        "category_normalized_disassembly",
                        "ssdeep_disassembly",
                        "tlsh_disassembly",
                        "ssdeep_fully_normalized",
                        "tlsh_fully_normalized",
                        "function_type",
                        "instruction_count",
                        "instruction_types",
                        "control_flow_count",
                        "memory_access_pattern",
                        "register_usage",
                        "data_references_count",
                        # 'opcode_frequency_vector',
                        # 'api_calls_vector',
                        # 'minhash_signature',
                        # 'pic_hash',
                        "max_block_size",
                        "num_calls",
                        "stack_size",
                        # 'instruction_type_ratios',
                        # 'instruction_embedding',
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "FixedString(64)",
                        "FixedString(64)",
                        "FixedString(64)",
                        "String",
                        "String",
                        "String",
                        "String",
                        "Nullable(String)",
                        "Nullable(FixedString(72))",
                        "Nullable(String)",
                        "Nullable(FixedString(72))",
                        "Enum8('USER'=1, 'LIBRARY'=2, 'THUNK'=3, 'EXTERNAL'=4, 'UNKNOWN'=5)",
                        "UInt32",
                        "Array(LowCardinality(String))",
                        "UInt32",
                        "Array(LowCardinality(String))",
                        "Array(LowCardinality(String))",
                        "UInt32",
                        # 'Array(Float32)',
                        # 'Array(Float32)',
                        # 'Array(UInt64)',
                        # 'Nullable(FixedString(16))',
                        "Nullable(UInt32)",
                        "Nullable(UInt32)",
                        "Nullable(Int32)",
                        # 'Array(Float32)',
                        # 'Array(Float32)',
                        "DateTime64(3, 'UTC')",
                    ],
                },
                "disassembled_refs": {
                    "table": "disassembled_functions_references",
                    "data": [
                        [
                            self.analysis_results["sha256"],
                            f["disassembled_content_hash"],
                            f["fully_normalized_content_hash"],
                            f["api_normalized_content_hash"],
                            f["category_normalized_content_hash"],
                            f["disassembled_function_name"],
                            f["disassembled_function_address"],
                            now,
                        ]
                        for f in self.analysis_results["disassembled"]
                    ],
                    "column_names": [
                        "sha256",
                        "disassembled_content_hash",
                        "fully_normalized_content_hash",
                        "api_normalized_content_hash",
                        "category_normalized_content_hash",
                        "disassembled_function_name",
                        "disassembled_function_address",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "FixedString(64)",
                        "FixedString(64)",
                        "FixedString(64)",
                        "FixedString(64)",
                        "LowCardinality(String)",
                        "String",
                        "DateTime64(3, 'UTC')",
                    ],
                },
                "cfg_blocks": {
                    "table": "cfg_blocks",
                    "data": [
                        [
                            b["block_id"],
                            self.analysis_results["sha256"],
                            b["function_address"],
                            b["block_start_address"],
                            b["block_end_address"],
                            b["block_size"],
                            b["block_instructions"],
                            b["fully_normalized_instructions"],
                            b["api_normalized_instructions"],
                            b["category_normalized_instructions"],
                            b.get("predecessor_blocks", []),
                            b.get("successor_blocks", []),  # Use empty array as default
                            b.get("is_entry_block", False),
                            b.get("is_exit_block", False),
                            b.get("branch_type", "UNKNOWN"),
                            b.get("referenced_constants", []),
                            b.get("sign", 1),  # Use 1 as default for sign
                            now,
                        ]
                        for b in self.analysis_results["cfg"]
                    ],
                    "column_names": [
                        "block_id",
                        "sha256",
                        "function_address",
                        "block_start_address",
                        "block_end_address",
                        "block_size",
                        "block_instructions",
                        "fully_normalized_instructions",
                        "api_normalized_instructions",
                        "category_normalized_instructions",
                        "predecessor_blocks",
                        "successor_blocks",
                        "is_entry_block",
                        "is_exit_block",
                        "branch_type",
                        "referenced_constants",
                        "sign",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "FixedString(64)",
                        "String",
                        "String",
                        "String",
                        "UInt32",
                        "String",
                        "Nullable(String)",
                        "Nullable(String)",
                        "Nullable(String)",
                        "Array(String)",
                        "Array(String)",
                        "Bool",
                        "Bool",
                        "Enum8('DIRECT'=1, 'CONDITIONAL'=2, 'CALL'=3, 'RETURN'=4, 'FALLTHROUGH'=5, 'UNKNOWN'=6)",
                        "Array(String)",
                        "Int8",
                        "DateTime64(3, 'UTC')",
                    ],
                },
                "function_analysis_errors": {
                    "table": "function_analysis_errors",
                    "data": [
                        [
                            self.analysis_results["sha256"],
                            f["function_name"],
                            f["function_address"],
                            f["error_location"],
                            f.get(
                                "error_message", ""
                            ),  # it could be empty, how to handle it?
                            f.get("error_details", ""),
                            f.get("error_type", "unknown"),
                            md5(
                                f"{f['error_message']}{f['function_name']}{f['function_address']}{f['error_location']}".encode()
                            ).hexdigest(),
                            "new",
                            now,
                        ]
                        for f in self.analysis_results["errors"]
                    ],
                    "column_names": [
                        "sha256",
                        "function_name",
                        "function_address",
                        "error_location",
                        "error_message",
                        "error_details",
                        "error_type",
                        "error_hash",
                        "status",
                        "analysis_date",
                    ],
                    "column_type_names": [
                        "FixedString(64)",
                        "Nullable(String)",
                        "String",
                        "LowCardinality(String)",
                        "Nullable(String)",
                        "Nullable(String)",
                        "Nullable(String)",
                        "FixedString(32)",
                        "Enum8('new'=1, 'investigating'=2, 'fixed'=3, 'wontfix'=4)",
                        "DateTime64(3, 'UTC')",
                    ],
                },
            }

    def tag(self) -> str:
        """Return the tag for this extractor."""
        return Tag.DECOMPILED.value

    def get_clickhouse_table(self) -> str:
        """Not used directly as we're handling multiple tables."""
        pass