Hanchuan Xu

from dataclasses import asdict
import inspect
import json
import subprocess
import magic
from magika import Magika
from typing import Any, Dict, List, Tuple
from datetime import datetime, timezone

from redb.extractors.enum import Tag
from redb.models.dataclasses import CAPA
from redb.extractors.extractor import Extractor
from dotenv import load_dotenv
import os

load_dotenv(override=True)

class CAPAExtractor(Extractor):

    def __init__(
        self,
        filepath,
        log,
        exporters=None,
        index_prefix=None,
        elastic_index=None,
        known_benign=False,
        known_malicious=False,
    ):
        super().__init__(
            filepath, 
            log, 
            exporters,
            index_prefix, 
            elastic_index, 
            known_benign, 
            known_malicious
        )
        self.capa = None
        self.elastic_index = self.index_prefix + "-capa"

    def _replace_empty_keys(self, dictionary, replacement="default_empty_key"):
        """
        Replace empty keys in a dictionary with a specified replacement.

        Args:
        dictionary (dict): The input dictionary
        replacement (str): The replacement for empty keys (default: 'empty_key')

        Returns:
        dict: A new dictionary with empty keys replaced
        """
        # tmp_dict = {}
        # for k, v in dictionary.items():
        #     if k == '':
        #         tmp_dict[replacement] = dictionary[k]
        #     else:
        #         tmp_dict[k] = dictionary[k]
        # return tmp_dict
        return {(replacement if k == "" else k): v for k, v in dictionary.items()}

    def _extract_capa(self):
        self.log.debug(inspect.currentframe().f_code.co_name)
        TIMEOUT = int(os.getenv("CAPA_TIMEOUT")) # 5 minutes in seconds
        capa_command = [os.getenv("CAPA_PATH"), "-j", self.filepath] # COMMENT FOR TESTING ON MAC

        import signal

        # try:
        #     result = subprocess.run(
        #         capa_command, capture_output=True, text=True, check=True, timeout=TIMEOUT
        #     )
        # except subprocess.TimeoutExpired:
        #     self.log.error(f"The capa command timed out after {TIMEOUT} seconds")
        #     return {}
        # except subprocess.CalledProcessError as e:
        #     self.log.error(f"Error running capa: {e}")
        #     return {}

        # try:
        #     capa_output = json.loads(result.stdout)
        # except json.JSONDecodeError as e:
        #     self.log.error("Error parsing capa output: {e}")
        #     return {}

        # Create a process group for easier termination of all child processes
        try:
            # Start the process in a new process group
            process = subprocess.Popen(
                capa_command,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                preexec_fn=os.setsid  # Use os.setsid() to create a new process group
            )
            
            # Implement timeout handling manually
            try:
                stdout, stderr = process.communicate(timeout=TIMEOUT)
                # Process completed within timeout
                if process.returncode != 0:
                    self.log.error(f"Error running capa, return code: {process.returncode}, stderr: {stderr}")
                    return {}
            except subprocess.TimeoutExpired:
                # Kill the entire process group on timeout
                self.log.warning(f"The capa command timed out after {TIMEOUT} seconds, terminating process group")
                
                try:
                    # Send SIGTERM to the entire process group
                    os.killpg(process.pid, signal.SIGTERM)
                    
                    # Give it a moment to terminate gracefully
                    try:
                        process.wait(timeout=3)
                    except subprocess.TimeoutExpired:
                        # If it's still running after 3 seconds, send SIGKILL
                        self.log.warning("Process didn't terminate with SIGTERM, sending SIGKILL")
                        os.killpg(process.pid, signal.SIGKILL)
                        
                    process.wait()  # Make sure process resources are fully cleaned up
                except (ProcessLookupError, OSError) as e:
                    self.log.warning(f"Error while killing process: {e}")
                    
                return {}
                
            try:
                capa_output = json.loads(stdout)
            except json.JSONDecodeError as e:
                self.log.error(f"Error parsing capa output: {e}")
                return {}

            capa_dump = {}  # dictionalry of capabilities
            capabilities = set()  # list of capabilities
            namespaces = set()  # list of namespaces
            attack_dump = {}  # dictionary of tuples [tactic](technique, technique_id)
            tactics = set()  # list of tactics
            techniques = set()  # list of techniques
            techniques_id = set()  # list of techniques_id
            mbc_dump = {}  # dictionary of tuples [objective](behavior, behavior_id)
            mbc_objectives = set()  # list of objectives
            mbc_behaviors = set()  # list of behaviors
            mbc_behaviors_id = set()  # list of behaviors_id

            if isinstance(capa_output, dict) and "rules" in capa_output:
                rules = capa_output["rules"]
                if isinstance(rules, dict):
                    for rule_name, rule_data in rules.items():
                        if isinstance(rule_data, dict) and "meta" in rule_data:
                            meta = rule_data["meta"]
                            if isinstance(meta, dict):
                                # Process capabilities
                                namespace = meta.get("namespace", "").lower()
                                # if namespace != 'lib':
                                if namespace not in capa_dump:
                                    capa_dump[namespace] = []
                                    namespaces.add(namespace)
                                capa_dump[namespace].append(rule_name)
                                capabilities.add(rule_name)

                                # Process ATTACK information
                                if "attack" in meta:
                                    for attack_entry in meta["attack"]:
                                        tactic = attack_entry.get("tactic", "")
                                        technique = attack_entry.get("technique", "")
                                        id = attack_entry.get("id", "")
                                        if tactic and technique:
                                            if tactic in attack_dump:
                                                attack_dump[tactic].append((technique, id))
                                            else:
                                                attack_dump[tactic] = [(technique, id)]
                                            tactics.add(tactic)
                                            techniques.add(technique)
                                            techniques_id.add(id)

                                # Process MBC information
                                if "mbc" in meta:
                                    for mbc_entry in meta["mbc"]:
                                        objective = mbc_entry.get("objective", "")
                                        behavior = mbc_entry.get("behavior", "")
                                        id = mbc_entry.get("id", "")
                                        if objective and behavior:
                                            if objective in mbc_dump:
                                                mbc_dump[objective].append((behavior, id))
                                            else:
                                                mbc_dump[objective] = [(behavior, id)]
                                            mbc_objectives.add(objective)
                                            mbc_behaviors.add(behavior)
                                            mbc_behaviors_id.add(id)

            # return {
            #     'capabilities_all': json.dumps(capabilities_all, indent=2),
            #     'capabilities': list(capabilities),
            #     'namespaces': list(namespaces),
            #     'attack': json.dumps(attack, indent=2),
            #     'tactics': list(tactics),
            #     'techniques': list(techniques),
            #     'techniques_id': list(techniques_id),
            #     'mbc': json.dumps(mbc, indent=2),
            #     'mbc_objectives': list(mbc_objectives),
            #     'mbc_behaviors': list(mbc_behaviors),
            #     'mbc_behaviors_id': list(mbc_behaviors_id)
            # }

            capa_dump = self._replace_empty_keys(capa_dump)
            attack_dump = self._replace_empty_keys(attack_dump)
            mbc_dump = self._replace_empty_keys(mbc_dump)
            self.capa = CAPA(
                json.dumps(capa_dump, indent=2),
                list(capabilities),
                list(namespaces),
                json.dumps(attack_dump, indent=2),
                list(tactics),
                list(techniques),
                list(techniques_id),
                json.dumps(mbc_dump, indent=2),
                list(mbc_objectives),
                list(mbc_behaviors),
                list(mbc_behaviors_id),
            )
            self.log.debug(f"CAPA dump: {self.capa})")
            
        except Exception as e:
            self.log.error(f"Unexpected error in CAPA extraction: {str(e)}")
            # Try to clean up any process if possible
            if 'process' in locals() and process.poll() is None:
                try:
                    os.killpg(process.pid, signal.SIGKILL)
                    process.wait()
                except:
                    pass
            return {}

    def prepare_export_data(self, exporter_type: str) -> Any:
        if exporter_type == "ElasticsearchExporter":
            return self.capa
        elif exporter_type == "ClickHouseExporter":
            current_time = datetime.now(timezone.utc)
            
            # Prepare data for multiple tables
            tables_data = {
                'multi_table': True,  # Flag for multi-table export
                
                # Raw data table
                'raw': {
                    'table': 'redb_capa',
                    'data': [[
                        self.sha256,
                        self.md5,
                        self.sha1,
                        current_time,
                        self.capa.capa_dump,  # capa_dump
                        self.capa.attack_dump,            # attack_dump
                        self.capa.mbc_dump                # mbc_dump
                    ]],
                    'column_names': [
                        'sha256', 'md5', 'sha1', 'analysis_date',
                        'capa_dump', 'attack_dump', 'mbc_dump'
                    ],
                    'column_type_names': [
                        'FixedString(64)', 'FixedString(32)', 'FixedString(40)',
                        'DateTime64(3, \'UTC\')',
                        'JSON', 'JSON', 'JSON'
                    ]
                },
                
                # Capabilities table
                'capabilities': {
                    'table': 'redb_capa_capabilities',
                    'data': [
                        [self.sha256, self.md5, self.sha1, current_time, namespace, capability]
                        for namespace, capabilities in json.loads(self.capa.capa_dump).items()
                        for capability in capabilities
                    ],
                    'column_names': [
                        'sha256', 'md5', 'sha1', 'analysis_date',
                        'namespace', 'capability'
                    ],
                    'column_type_names': [
                        'FixedString(64)', 'FixedString(32)', 'FixedString(40)',
                        'DateTime64(3, \'UTC\')',
                        'LowCardinality(String)', 'LowCardinality(String)'
                    ]
                },
                
                # MITRE ATT&CK table
                'attack': {
                    'table': 'redb_capa_attack',
                    'data': [
                        [self.sha256, self.md5, self.sha1, current_time, 
                         tactic, technique[0], technique[1]]  # technique[0] is the name, technique[1] is the ID
                        for tactic, techniques in json.loads(self.capa.attack_dump).items()
                        for technique in techniques
                    ],
                    'column_names': [
                        'sha256', 'md5', 'sha1', 'analysis_date',
                        'tactic', 'technique', 'technique_id'
                    ],
                    'column_type_names': [
                        'FixedString(64)', 'FixedString(32)', 'FixedString(40)',
                        'DateTime64(3, \'UTC\')',
                        'LowCardinality(String)', 'LowCardinality(String)', 
                        'LowCardinality(String)'
                    ]
                },
                
                # MBC table
                'mbc': {
                    'table': 'redb_capa_mbc',
                    'data': [
                        [self.sha256, self.md5, self.sha1, current_time,
                         objective, behavior[0], behavior[1]]  # behavior[0] is the name, behavior[1] is the ID
                        for objective, behaviors in json.loads(self.capa.mbc_dump).items()
                        for behavior in behaviors
                    ],
                    'column_names': [
                        'sha256', 'md5', 'sha1', 'analysis_date',
                        'objective', 'behavior', 'behavior_id'
                    ],
                    'column_type_names': [
                        'FixedString(64)', 'FixedString(32)', 'FixedString(40)',
                        'DateTime64(3, \'UTC\')',
                        'LowCardinality(String)', 'LowCardinality(String)',
                        'LowCardinality(String)'
                    ]
                }
            }
            
            return tables_data

    def get_clickhouse_table(self) -> str:
        return "redb_capa"  # Return the main table name

    def extract(self):
        self.log.debug(inspect.currentframe().f_code.co_name)
        try:
            self._extract_capa()
            return self.capa  # Return the extracted data instead of exporting directly
        except Exception as e:
            self.log.error(f"Error extracting CAPA: {e}")
            return None

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