Hangbing Lv

import inspect
import binascii
from datetime import datetime, timezone
from typing import Any, List, Dict

from elftools.elf.elffile import ELFFile
from elftools.common.exceptions import ELFError

from redb.extractors.enum import Tag
from redb.extractors.elf_extractor import ELFExtractor
from redb.models.dataclasses import ELFNote


class ELFNotesExtractor(ELFExtractor):

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

    def _get_note_type_string(self, note_type: int, note_name: str) -> str:
        """Convert note type number to human-readable string."""

        # GNU-specific note types
        if note_name == "GNU":
            gnu_types = {
                1: "NT_GNU_ABI_TAG",
                2: "NT_GNU_HWCAP",
                3: "NT_GNU_BUILD_ID",
                4: "NT_GNU_GOLD_VERSION",
                5: "NT_GNU_PROPERTY_TYPE_0"
            }
            return gnu_types.get(note_type, f"NT_GNU_UNKNOWN_{note_type}")

        # Generic note types
        generic_types = {
            1: "NT_PRSTATUS",
            2: "NT_FPREGSET",
            3: "NT_PRPSINFO",
            4: "NT_TASKSTRUCT",
            5: "NT_AUXV",
            6: "NT_PSTATUS",
            7: "NT_FPREGS",
            8: "NT_PSINFO",
            9: "NT_PRCRED",
            10: "NT_UTSNAME",
            11: "NT_LWPSTATUS",
            12: "NT_LWPSINFO",
            13: "NT_PRFPXREG"
        }

        return generic_types.get(note_type, f"NT_UNKNOWN_{note_type}")

    def _format_note_description(self, note_desc, note_type: int, note_name: str) -> str:
        """Format note description based on type for human readability."""
        try:
            if not note_desc:
                return ""

            # Handle build ID specifically (common case)
            if note_name == "GNU" and note_type == 3:  # NT_GNU_BUILD_ID
                if isinstance(note_desc, bytes):
                    return binascii.hexlify(note_desc).decode('ascii')
                return str(note_desc)

            # Handle ABI tag
            if note_name == "GNU" and note_type == 1:  # NT_GNU_ABI_TAG
                if isinstance(note_desc, bytes) and len(note_desc) >= 16:
                    # ABI tag contains OS, major, minor, subminor
                    import struct
                    try:
                        os_val, major, minor, subminor = struct.unpack('<IIII', note_desc[:16])
                        os_names = {0: "Linux", 1: "GNU", 2: "Solaris", 3: "FreeBSD"}
                        os_name = os_names.get(os_val, f"OS_{os_val}")
                        return f"{os_name} {major}.{minor}.{subminor}"
                    except:
                        pass

            # For binary data, convert to hex
            if isinstance(note_desc, bytes):
                # Limit size for very large descriptions
                if len(note_desc) > 256:
                    return binascii.hexlify(note_desc[:256]).decode('ascii') + "..."
                return binascii.hexlify(note_desc).decode('ascii')

            # For string data
            if isinstance(note_desc, str):
                return note_desc

            # Fallback
            return str(note_desc)

        except Exception as e:
            self.log.error(f"Error formatting note description: {e}")
            return str(note_desc) if note_desc else ""

    def _extract_note_data(self, note, section_name: str) -> ELFNote:
        """Extract data from a single note entry."""
        try:
            # Get note properties
            note_name = note.get('n_name', '').rstrip('\x00') if note.get('n_name') else ""
            note_type_raw = note.get('n_type', 0)
            note_desc_raw = note.get('n_desc', b'')

            # Handle note_type - pyelftools may return string or int
            if isinstance(note_type_raw, str):
                # pyelftools returned the type as a string like 'NT_GNU_BUILD_ID'
                note_type_str = note_type_raw
                # Map known string types to integers
                note_type_map = {
                    'NT_GNU_ABI_TAG': 1,
                    'NT_GNU_HWCAP': 2,
                    'NT_GNU_BUILD_ID': 3,
                    'NT_GNU_GOLD_VERSION': 4,
                    'NT_GNU_PROPERTY_TYPE_0': 5,
                    'NT_PRSTATUS': 1,
                    'NT_FPREGSET': 2,
                    'NT_PRPSINFO': 3,
                    'NT_TASKSTRUCT': 4,
                    'NT_AUXV': 5,
                    'NT_PSTATUS': 6,
                    'NT_FPREGS': 7,
                    'NT_PSINFO': 8,
                    'NT_PRCRED': 9,
                    'NT_UTSNAME': 10,
                    'NT_LWPSTATUS': 11,
                    'NT_LWPSINFO': 12,
                    'NT_PRFPXREG': 13,
                }
                note_type = note_type_map.get(note_type_raw, 0)
            else:
                note_type = note_type_raw
                # Get human-readable type string
                note_type_str = self._get_note_type_string(note_type, note_name)

            # Format description
            note_desc = self._format_note_description(note_desc_raw, note_type, note_name)

            return ELFNote(
                note_name=note_name,
                note_type=note_type,
                note_type_str=note_type_str,
                note_desc=note_desc,
                note_section=section_name
            )

        except Exception as e:
            self.log.error(f"Error extracting note data: {e}")
            return None

    def _extract_notes_from_sections(self, elf) -> List[Dict]:
        """Extract notes from note sections."""
        notes = []

        try:
            # Look for note sections
            for section in elf.iter_sections():
                if (section.name and
                    section.name.startswith('.note') and
                    hasattr(section, 'iter_notes')):

                    section_name = section.name
                    try:
                        for note in section.iter_notes():
                            note_data = self._extract_note_data(note, section_name)
                            if note_data:
                                notes.append(note_data)
                    except Exception as e:
                        self.log.debug(f"Could not process notes in section {section_name}: {e}")

        except Exception as e:
            self.log.error(f"Error extracting notes from sections: {e}")

        return notes

    def _extract_notes_from_segments(self, elf) -> List[Dict]:
        """Extract notes from PT_NOTE segments."""
        notes = []

        try:
            # Look for PT_NOTE segments
            for segment in elf.iter_segments():
                if segment.header.get('p_type') == 'PT_NOTE':
                    segment_name = f"PT_NOTE_segment_{segment.header.get('p_offset', 0)}"

                    try:
                        if hasattr(segment, 'iter_notes'):
                            for note in segment.iter_notes():
                                note_data = self._extract_note_data(note, segment_name)
                                if note_data:
                                    notes.append(note_data)
                    except Exception as e:
                        self.log.debug(f"Could not process notes in segment: {e}")

        except Exception as e:
            self.log.error(f"Error extracting notes from segments: {e}")

        return notes

    def tag(self):
        return Tag.ELF_NOTES.value if hasattr(Tag, 'ELF_NOTES') else "elf_notes"

    def extract(self):
        try:
            self.log.debug(inspect.currentframe().f_code.co_name)

            def extract_data(elf):
                all_notes = []

                # Extract notes from note sections
                section_notes = self._extract_notes_from_sections(elf)
                all_notes.extend(section_notes)

                # Extract notes from PT_NOTE segments
                segment_notes = self._extract_notes_from_segments(elf)
                all_notes.extend(segment_notes)

                # Remove duplicates (same note might appear in section and segment)
                unique_notes = []
                seen_notes = set()
                for note in all_notes:
                    note_key = (note.note_name, note.note_type, note.note_desc)
                    if note_key not in seen_notes:
                        seen_notes.add(note_key)
                        unique_notes.append(note)

                return unique_notes

            if not self._is_elf_file():
                return None

            result = self._with_elf_file(extract_data)
            if result is None:
                return None

            self.elf_notes = result
            return self.elf_notes

        except Exception as e:
            self.log.error(f"Error extracting ELF notes {self.hash.sha256}: {e}")
            return None

    def prepare_export_data(self, exporter_type: str) -> Any:
        self.log.debug(inspect.currentframe().f_code.co_name)

        if exporter_type == "ElasticsearchExporter":
            return self.elf_notes
        elif exporter_type == "ClickHouseExporter":
            try:
                # Return valid empty structure if no notes found
                # None is reserved for actual errors

                # Prepare data arrays for all notes
                data = []
                current_time = datetime.now(timezone.utc)
                for note in self.elf_notes:
                    row = [
                        self.sha256,
                        self.md5,
                        self.sha1,
                        note.note_name,
                        note.note_type,
                        note.note_type_str,
                        note.note_desc,
                        note.note_section,
                        current_time
                    ]
                    data.append(row)

                column_names = [
                    'sha256', 'md5', 'sha1',
                    'note_name', 'note_type', 'note_type_str',
                    'note_desc', 'note_section',
                    'analysis_date'
                ]

                if not data:
                    return None

                column_type_names = [
                    'FixedString(64)', 'FixedString(32)', 'FixedString(40)',
                    'LowCardinality(String)', 'UInt32', 'LowCardinality(String)',
                    'String CODEC(ZSTD(3))', 'LowCardinality(String)',
                    'DateTime64(3, \'UTC\')'
                ]

                return (data, column_names, column_type_names)

            except Exception as e:
                self.log.error(f"Error preparing export data: {e}")
                raise

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