Xiao-Sheng Si

import inspect
from pprint import pprint
import subprocess
import json
from typing import Any
from datetime import datetime, timezone
import os
from dotenv import load_dotenv

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

load_dotenv(override=True)

class DIEExtractor(Extractor):

    def __init__(
        self,
        filepath,
        log,
        exporters=None,
        index_prefix=None,
        elastic_index=None,
        known_benign=False,
        known_malicious=False,
        precomputed_hashes=None,
    ):
        super().__init__(
            filepath, log, exporters, index_prefix, elastic_index, known_benign, known_malicious,
            precomputed_hashes=precomputed_hashes
        )
        self.log.debug(inspect.currentframe().f_code.co_name)
        self.die_info = None
        self.die_info_dict = {}
        self.elastic_index = self.index_prefix + "-die"

    def _recursive_entry(self, die_dict, master_key):
        self.log.debug(inspect.currentframe().f_code.co_name)
        if master_key:
            self.die_info_dict[master_key] = {}
        else:
            self.die_info_dict = {}
        for value in die_dict:
            if "type" in value:
                type_key = value["type"].lower().replace(" ", "_")
                name = value.get("name", "")
                version = f"({value.get('version')})" if value.get("version") else ""
                info = f"[{value.get('info')}]" if value.get("info") else ""

                if master_key:
                    self.die_info_dict[master_key][type_key] = f"{name}"
                    self.die_info_dict[master_key][f'{type_key}(full)'] = f"{name}{version}{info}"
                else:
                    self.die_info_dict[type_key] = f"{name}"
                    self.die_info_dict[f'{type_key}(full)'] = f"{name}{version}{info}"

            elif "parentfilepart" in value:
                child_key = (
                    value["parentfilepart"].lower().replace(" ", "_")
                    + "."
                    + value["filetype"].lower().replace(" ", "_")
                )
                if master_key:
                    self._recursive_entry(value["values"], f"{master_key}.{child_key}")
                else:
                    self._recursive_entry(value["values"], f"{child_key}")

    def _extract_dieinfo(self):
        """
        Execute a command-line binary with arguments and parse its JSON output.

        :param command: The command or path to the binary to execute
        :param args: Additional arguments to pass to the command
        :return: Parsed JSON output as a Python object
        """
        self.log.debug(inspect.currentframe().f_code.co_name)

        # Construct the full command
        # command = "nfdc" # UNCOMMENT FOR PROD
        # command = "/Users/p4c0/_tools/NFD.app/Contents/MacOS/nfdc" # COMMENT FOR TESTING ON MAC
        command = os.getenv("DIE_PATH")
        args = ["-durj", self.filepath]
        full_command = [command] + list(args)
        TIMEOUT = int(os.getenv("DIE_TIMEOUT", "180"))

        try:
            # Execute the command and capture its output
            result = subprocess.run(
                full_command,
                capture_output=True,
                text=True,
                check=True,
                timeout=TIMEOUT,
            )

            # Parse the JSON output
            nfdc_output = json.loads(result.stdout)

            # Extract the DIE information from the json output
            for die_entry in nfdc_output["detects"]:
                if die_entry["parentfilepart"] == "Header":
                    master_key = (
                        die_entry["parentfilepart"].lower().replace(" ", "_")
                        + "."
                        + die_entry["filetype"].lower().replace(" ", "_")
                    )
                    self._recursive_entry(die_entry["values"], None)

            # pprint(json.dumps(self.die_info_dict, indent=2)) #debug
            self.die_info = DIEinfo(result.stdout, self.die_info_dict)
            self.log.debug(f"NFDC-DIE JSON dump: todo")
        except subprocess.TimeoutExpired:
            self.log.error(f"The DIE command timed out after {TIMEOUT} seconds")
            return None
        except subprocess.CalledProcessError as e:
            self.log.error(f"Error executing DIE command: {e}")
            self.log.error(f"Command output (stderr): {e.stderr}")
            return None
        except json.JSONDecodeError as e:
            self.log.error(f"Error parsing DIE JSON output: {e}")
            self.log.error(f"Raw output: {result.stdout}")
            return None

    def prepare_export_data(self, exporter_type: str) -> Any:
        if exporter_type == "ElasticsearchExporter":
            return self.die_info
        elif exporter_type == "ClickHouseExporter":
            # Convert DIE info to JSON string
            die_info_json = json.dumps(self.die_info_dict)
            
            data = [[
                self.sha256,
                self.md5,
                self.sha1,
                die_info_json,
                datetime.now(timezone.utc)
            ]]
            
            column_names = [
                'sha256', 'md5', 'sha1', 'die_info', 'analysis_date'
            ]
            
            column_type_names = [
                'String', 'String', 'String', 'JSON', 'DateTime64(3, \'UTC\')'
            ]
            
            return (data, column_names, column_type_names)

    def get_clickhouse_table(self) -> str:
        return "redb_die"

    def extract(self):
        self.log.debug(inspect.currentframe().f_code.co_name)
        try:
            self._extract_dieinfo()
            
            # Check if there's a packer in the DIE results
            is_packed = False
            if self.die_info_dict:
                # Check if 'packer' exists in the DIE results
                is_packed = bool(self.die_info_dict.get('packer'))
            
            return self.die_info  # Return the extracted data instead of exporting directly
        except Exception as e:
            self.log.error(f"Error extracting DIE information: {e}")
            return None

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