Xiaobo Xue

from hashlib import md5, sha1, sha256
import inspect
from datetime import datetime, timezone
from typing import Any

import magic
from magika import Magika

from redb.extractors.enum import Tag
from redb.extractors.pe_extractor import PEExtractor
from redb.models.dataclasses import PEOverlay


class PEOverlayExtractor(PEExtractor):

    def __init__(
        self,
        filepath,
        log,
        exporters=None,
        index_prefix=None,
        elastic_index=None,
        known_benign=False,
        known_malicious=False,
        pe=None,
    ):
        super().__init__(
            filepath,
            log,
            exporters,
            index_prefix,
            elastic_index,
            known_benign,
            known_malicious,
            pe,
        )
        self.elastic_index = self.index_prefix + "-pe_overlay"
        self.log.debug(inspect.currentframe().f_code.co_name)

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

    def _extract_pe_overlay(self):
        self.log.debug(inspect.currentframe().f_code.co_name)
        overlay = self.pe.get_overlay()
        if overlay:
            return PEOverlay(
                _id=sha256(overlay).hexdigest(),
                overlay_size=len(overlay),
                overlay_entropy=self.calculate_entropy(overlay),
                overlay_offset=self.pe.get_overlay_data_start_offset(),
                overlay_mimetype=magic.from_buffer(overlay, mime=True),
                overlay_type=magic.from_buffer(overlay),
                overlay_magika=Magika().identify_bytes(overlay).output.ct_label,
                overlay_sha256=sha256(overlay).hexdigest(),
                overlay_sha1=sha1(overlay).hexdigest(),
                overlay_md5=md5(overlay).hexdigest(),
            )
        return None

    def extract(self):
        try:
            self.log.debug(inspect.currentframe().f_code.co_name)
            overlay = self._extract_pe_overlay()
            # self.export_to_elastic([overlay])  # Let the exporters handle this
            return overlay
        except Exception as e:
            self.log.error(f"Error extracting PE overlay: {e}")
            return None

    def prepare_export_data(self, exporter_type: str) -> Any:
        if exporter_type == "ElasticsearchExporter":
            return self.extract()
        elif exporter_type == "ClickHouseExporter":
            overlay = self.extract()
            if overlay is None:
                return None
            
            data = []
            current_time = datetime.now(timezone.utc)
            
            data.append([
                self.sha256,              # sha256
                self.md5,                 # md5
                self.sha1,                # sha1
                overlay.overlay_size,     # overlay_size
                overlay.overlay_entropy,  # overlay_entropy
                overlay.overlay_offset,   # overlay_offset
                overlay.overlay_mimetype, # overlay_mimetype
                overlay.overlay_type,     # overlay_type
                overlay.overlay_magika,   # overlay_magika
                overlay.overlay_sha256,   # overlay_sha256
                overlay.overlay_sha1,     # overlay_sha1
                overlay.overlay_md5,      # overlay_md5
                current_time             # analysis_date
            ])
            
            column_names = [
                'sha256', 'md5', 'sha1', 'overlay_size', 'overlay_entropy',
                'overlay_offset', 'overlay_mimetype', 'overlay_type', 'overlay_magika',
                'overlay_sha256', 'overlay_sha1', 'overlay_md5',
                'analysis_date'
            ]
            
            if not data:
                return None

            column_type_names = [
                'FixedString(64)', 'FixedString(32)', 'FixedString(40)',
                'UInt64', 'Float64', 'UInt64',
                'LowCardinality(String)', 'LowCardinality(String)', 'LowCardinality(String)',
                'FixedString(64)', 'FixedString(40)', 'FixedString(32)',
                'DateTime64(3, \'UTC\')'
            ]

            return (data, column_names, column_type_names)

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