K. Thangavel

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"