Kaihua Cao

from dataclasses import asdict
import inspect
import os
import magic
from magika import Magika
from datetime import datetime, timezone
from typing import Any, List, Tuple

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


class BasicPropertiesExtractor(Extractor):

    def __init__(
        self,
        filepath,
        log,
        exporters=None,
        index_prefix=None,
        elastic_index=None,
        known_benign=False,
        known_malicious=False,
        parent_sha256=None,
        precomputed_hashes=None,
        is_fat=None,
        child_sha256=None,
        child_architecture=None,
        child_filetype=None,
        first_seen=None,
    ):
        super().__init__(
            filepath,
            log,
            exporters=exporters,
            index_prefix=index_prefix,
            elastic_index=elastic_index,
            known_benign=known_benign,
            known_malicious=known_malicious,
            precomputed_hashes=precomputed_hashes,
        )
        self.basic_properties = None
        self.elastic_index = self.index_prefix + "-basic_properties"
        self.is_packed = None
        self.parent_sha256 = parent_sha256  # For FAT Mach-O slices, points to container hash
        self.is_fat = is_fat  # True for FAT Mach-O containers, None otherwise
        self.child_sha256 = child_sha256  # SHA256 hashes of children (FAT slices, zip contents)
        self.child_architecture = child_architecture  # Architecture names for FAT Mach-O slices
        self.child_filetype = child_filetype  # Magika filetypes for children (useful for zip archives)
        self.first_seen = first_seen  # From catalog_samples, None for local files

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

    def _extract_basic_properties(self):
        self.log.debug(inspect.currentframe().f_code.co_name)
        filename = None  # Reserved for future use
        sample_name = os.path.basename(self.filepath)
        file_entropy = round(self.calculate_entropy(self.binary), 3)
        type_ = magic.from_buffer(self.binary)
        type_mime = magic.from_buffer(self.binary, mime=True)
        type_magika = Magika().identify_bytes(self.binary).output.label
        size = len(self.binary)
        self.basic_properties = BasicProperties(
            filename, sample_name, size, type_, type_mime, type_magika,
            file_entropy, self.is_packed, self.is_fat, self.child_sha256,
            self.child_architecture, self.child_filetype,
            first_seen=str(self.first_seen) if self.first_seen else None,
        )
        self.log.debug(f"Basic Properties dump: {asdict(self.basic_properties)}")

    def extract(self):
        self.log.debug(inspect.currentframe().f_code.co_name)
        try:
            self._extract_basic_properties()
            # self.export_to_elastic([self.basic_properties])
            return self.basic_properties
        except Exception as e:
            self.log.error(f"Extract basic properties error {self.hash.sha256} Exception: {e}")
            return None

    def prepare_export_data(self, exporter_type: str) -> Any:
        if exporter_type == "ElasticsearchExporter":
            return self.basic_properties
        elif exporter_type == "ClickHouseExporter":
            data = [[
                self.sha256,
                self.md5,
                self.sha1,
                self.parent_sha256,  # NULL for standalone/FAT, fat_hash for slices
                self.basic_properties.child_sha256 or [],  # SHA256 hashes of children
                self.basic_properties.child_architecture or [],  # Architecture names for FAT slices
                self.basic_properties.child_filetype or [],  # Magika filetypes for children
                self.basic_properties.is_fat,  # True for FAT Mach-O containers, NULL otherwise
                self.basic_properties.filename,
                self.basic_properties.sample_name,
                self.basic_properties.filesize,
                self.basic_properties.file_entropy,
                self.basic_properties.filetype,
                self.basic_properties.filetype_mime,
                self.basic_properties.filetype_magika,
                self.first_seen or datetime(1970, 1, 1, tzinfo=timezone.utc),  # From catalog_samples, epoch zero for local files (uses original datetime, not string)
                datetime.now(timezone.utc)
            ]]

            column_names = [
                'sha256', 'md5', 'sha1', 'parent_sha256', 'child_sha256', 'child_architecture', 'child_filetype',
                'is_fat', 'filename', 'sample_name',
                'filesize', 'file_entropy', 'filetype', 'filetype_mime',
                'filetype_magika', 'first_seen', 'analysis_date'
            ]

            column_type_names = [
                'String', 'String', 'String', 'Nullable(String)',
                'Array(FixedString(64))', 'Array(LowCardinality(String))', 'Array(LowCardinality(String))',
                'Nullable(UInt8)', 'Nullable(String)', 'String',
                'UInt64', 'Float64', 'LowCardinality(String)',
                'LowCardinality(String)', 'LowCardinality(String)',
                'DateTime64(3, \'UTC\')',
                'DateTime64(3, \'UTC\')'
            ]

            return data, column_names, column_type_names

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