Xavier Le Pallec

import inspect
import yara_x
from datetime import datetime, timezone
from signify.authenticode import SignedPEFile
from typing import Any, Dict, List, Tuple

from redb.extractors.enum import Tag
from redb.extractors.pe_extractor import PEExtractor
from redb.models.dataclasses import PECertificate, PESigner, PECodeSigningInfo


class PESignatureExtractor(PEExtractor):
    """
    RFC 3161 - Internet X.509 Public Key Infrastructure Time-Stamp Protocol (TSP)
    Section 2.4.2 defines the TSTInfo structure.
    https://tools.ietf.org/html/rfc3161#section-2.4.2

    Microsoft Authenticode Time Stamping Specification
    https://download.microsoft.com/download/9/c/5/9c5b2167-8017-4bae-9fde-d599bac8184a/Authenticode_PE.docx
    """

    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_signature"
        self.log.debug(inspect.currentframe().f_code.co_name)
        self.code_signing_info = None

    def tag(self):
        return [Tag.PE_CERTIFICATE.value, Tag.PE_SIGNER.value, Tag.PE_SIGNATURE.value]

    def _extract_certificate_info(self, certificate):
        self.log.debug(inspect.currentframe().f_code.co_name)

        # YARA-X returns datetime objects directly, not Unix timestamps
        not_before = certificate["not_before"]
        not_after = certificate["not_after"]

        # Handle both datetime objects and Unix timestamps for compatibility
        if isinstance(not_before, datetime):
            valid_from = not_before.strftime("%Y-%m-%d %H:%M:%S")
        else:
            valid_from = datetime.fromtimestamp(int(not_before), timezone.utc).strftime("%Y-%m-%d %H:%M:%S")

        if isinstance(not_after, datetime):
            valid_to = not_after.strftime("%Y-%m-%d %H:%M:%S")
        else:
            valid_to = datetime.fromtimestamp(int(not_after), timezone.utc).strftime("%Y-%m-%d %H:%M:%S")

        return PECertificate(
            certificate_serial_number=certificate["serial"],
            certificate_subject=certificate["subject"],
            certificate_issuer=certificate["issuer"],
            certificate_valid_from=valid_from,
            certificate_valid_to=valid_to,
            certificate_thumbprint=certificate["thumbprint"].upper(),
            certificate_algorithm=certificate["algorithm"],
        )

    def _extract_signer_info(self, signer):
        self.log.debug(inspect.currentframe().f_code.co_name)

        # YARA-X returns datetime objects directly, not Unix timestamps
        not_before = signer["not_before"]
        not_after = signer["not_after"]

        # Handle both datetime objects and Unix timestamps for compatibility
        if isinstance(not_before, datetime):
            valid_from = not_before.strftime("%Y-%m-%d %H:%M:%S")
        else:
            valid_from = datetime.fromtimestamp(int(not_before), timezone.utc).strftime("%Y-%m-%d %H:%M:%S")

        if isinstance(not_after, datetime):
            valid_to = not_after.strftime("%Y-%m-%d %H:%M:%S")
        else:
            valid_to = datetime.fromtimestamp(int(not_after), timezone.utc).strftime("%Y-%m-%d %H:%M:%S")

        return PESigner(
            signer_serial_number=signer["serial"],
            signer_subject=signer["subject"],
            signer_issuer=signer["issuer"],
            signer_valid_from=valid_from,
            signer_valid_to=valid_to,
            signer_thumbprint=signer["thumbprint"].upper(),
            signer_algorithm=signer["algorithm"],
        )

    def _extract_signature_info(self):
        self.log.debug(inspect.currentframe().f_code.co_name)
        pe_rules = """
            import "pe"
            rule is_pe {
                condition: 
                    pe.is_pe
            }
        """
        rule = yara_x.compile(pe_rules)
        results = rule.scan(self.binary)
        yara_output = results.module_outputs

        signature_info = yara_output["pe"]["signatures"][0]
        number_of_certificates = signature_info["number_of_certificates"]
        signature_verified = signature_info["verified"]
        self.code_signing_info = PECodeSigningInfo(
            _id=self.sha256,
            signature_verified=signature_verified,
            number_of_certificates=int(number_of_certificates),
        )

        keys = signature_info.keys()
        if "certificates" in keys:
            self.log.debug("Extracting PE Signing Certificates")
            certificates = signature_info["certificates"]
            certificates_list = []
            for cert in certificates:
                cert_info = self._extract_certificate_info(cert)
                certificates_list.append(cert_info)
            self.code_signing_info.x509_certificates = certificates_list

        if "signer_info" in keys:
            self.log.debug("Extracting PE Signing Signer Info")
            signers = signature_info["signer_info"]["chain"]
            signers_list = []
            for signer in signers:
                sig = self._extract_signer_info(signer)
                signers_list.append(sig)
            self.code_signing_info.signers = signers_list

        if "countersignatures" in keys:
            self.log.debug("Extracting PE Signing Countersignatures")
            countersignatures = signature_info["countersignatures"][0]["chain"]
            countersigners_list = []
            for csigner in countersignatures:
                csig = self._extract_signer_info(csigner)
                countersigners_list.append(csig)
            self.code_signing_info.countersigners = countersigners_list

        with open(self.filepath, "rb") as f:
            self.log.debug("Extracting PE Signing spcSpOpusInfo")
            sig_pefile = SignedPEFile(f)
            try:
                spc_opusinfo = {}
                for signed_data in sig_pefile.signed_datas:
                    spc_opusinfo["ProgramName"] = signed_data.signer_info.program_name
                    spc_opusinfo["MoreInfo"] = signed_data.signer_info.more_info
                    self.code_signing_info.spcSpOpusInfo = spc_opusinfo

                    if signed_data.signer_info.countersigner is not None:
                        stime = signed_data.signer_info.countersigner.signing_time
                        self.code_signing_info.date_signed = stime.strftime(
                            "%Y-%m-%d %H:%M:%S"
                        )
            except Exception as e:
                self.log.error(
                    f"Error while extracting spcSpOpusInfo or signing date: {e}"
                )

    def extract(self):
        try:
            self.log.debug(inspect.currentframe().f_code.co_name)
            if self._is_signed():
                self._extract_signature_info()
                # self.export_to_elastic([self.code_signing_info])
                return self.code_signing_info
            return None
        except Exception as e:
            self.log.error(f"Error extracting PE Signature: {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.extract()
    #     elif exporter_type == "ClickHouseExporter":
    #         signing_info = self.extract()
    #         if signing_info is None:
    #             return None
                
    #         current_time = datetime.now(timezone.utc)
            
    #         # Prepare data for redb_pe_signatures table
    #         signatures_data = [[
    #             self.sha256,                                    # sha256
    #             self.md5,                                       # md5
    #             self.sha1,                                      # sha1
    #             signing_info.signature_verified,                # signature_verified
    #             datetime.strptime(signing_info.date_signed, "%Y-%m-%d %H:%M:%S") if signing_info.date_signed else None,  # date_signed
    #             signing_info.number_of_certificates,            # number_of_certificates
    #             signing_info.spcSpOpusInfo.get('ProgramName', None) if signing_info.spcSpOpusInfo else None,  # program_name
    #             signing_info.spcSpOpusInfo.get('MoreInfo', None) if signing_info.spcSpOpusInfo else None,    # more_info
    #             current_time                                    # analysis_date
    #         ]]

    #         certificates_data = []
            
    #         # Process signers
    #         if signing_info.signers:
    #             for signer in signing_info.signers:
    #                 certificates_data.append(self._prepare_certificate_data(signer, 'signer', current_time))
            
    #         # Process countersigners
    #         if signing_info.countersigners:
    #             for csigner in signing_info.countersigners:
    #                 certificates_data.append(self._prepare_certificate_data(csigner, 'countersigner', current_time))
            
    #         # Process x509 certificates
    #         if signing_info.x509_certificates:
    #             for cert in signing_info.x509_certificates:
    #                 certificates_data.append(self._prepare_certificate_data(cert, 'chain', current_time))
            
    #         return {
    #             'multi_table': True,
    #             'signatures': {
    #                 'table': 'redb_pe_signatures',
    #                 'data': signatures_data,
    #                 'column_names': [
    #                     'sha256', 'md5', 'sha1', 'signature_verified', 'date_signed',
    #                     'number_of_certificates', 'spcSpOpusInfo_program_name', 'spcSpOpusInfo_more_info', 'analysis_date'
    #                 ],
    #                 'column_type_names': [
    #                     'FixedString(64)', 'FixedString(32)', 'FixedString(40)',
    #                     'Boolean', 'DateTime64(3, \'UTC\')', 'UInt8',
    #                     'Nullable(String)', 'Nullable(String)', 'DateTime64(3, \'UTC\')'
    #                 ]
    #             },
    #             'certificates': {
    #                 'table': 'redb_pe_certificates',
    #                 'data': certificates_data,
    #                 'column_names': [
    #                     'sha256', 'md5', 'sha1', 'certificate_type', 'certificate_algorithm',
    #                     'certificate_issuer', 'certificate_subject', 'certificate_serial_number',
    #                     'certificate_thumbprint', 'certificate_valid_from', 'certificate_valid_to',
    #                     'analysis_date'
    #                 ],
    #                 'column_type_names': [
    #                     'FixedString(64)', 'FixedString(32)', 'FixedString(40)',
    #                     'Enum(\'signer\', \'countersigner\', \'chain\')', 'LowCardinality(String)',
    #                     'LowCardinality(String)', 'LowCardinality(String)', 'String',
    #                     'String', 'DateTime64(3, \'UTC\')', 'DateTime64(3, \'UTC\')',
    #                     'DateTime64(3, \'UTC\')'
    #                 ]
    #             }
    #         }

    # def _prepare_certificate_data(self, cert, cert_type: str, analysis_time: datetime) -> List:
    #     """Helper method to prepare certificate data for ClickHouse"""
    #     self.log.debug(inspect.currentframe().f_code.co_name)
    #     prefix = "certificate_" if hasattr(cert, "certificate_algorithm") else "signer_"
        
    #     # Get serial number and clean it up
    #     serial_number = getattr(cert, f"{prefix}serial_number")
    #     if serial_number:
    #         serial_number = serial_number.replace(':', '')
        
    #     return [
    #         self.sha256,                          # sha256
    #         self.md5,                             # md5
    #         self.sha1,                            # sha1
    #         cert_type,                            # certificate_type
    #         getattr(cert, f"{prefix}algorithm"),   # certificate_algorithm
    #         getattr(cert, f"{prefix}issuer"),     # certificate_issuer
    #         getattr(cert, f"{prefix}subject"),    # certificate_subject
    #         serial_number,                        # certificate_serial_number
    #         getattr(cert, f"{prefix}thumbprint"),  # certificate_thumbprint
    #         datetime.strptime(getattr(cert, f"{prefix}valid_from"), "%Y-%m-%d %H:%M:%S"),  # certificate_valid_from
    #         datetime.strptime(getattr(cert, f"{prefix}valid_to"), "%Y-%m-%d %H:%M:%S"),    # certificate_valid_to
    #         analysis_time                         # analysis_date
    #     ]
    def prepare_export_data(self, exporter_type: str) -> Any:
        self.log.debug(inspect.currentframe().f_code.co_name)
        if exporter_type == "ElasticsearchExporter":
            return self.extract()
        elif exporter_type == "ClickHouseExporter":
            signing_info = self.extract()
            if signing_info is None:
                return None
                
            current_time = datetime.now(timezone.utc)
            
            # Prepare data for redb_pe_signatures table
            signatures_data = [[
                self.sha256,                                    # sha256
                self.md5,                                       # md5
                self.sha1,                                      # sha1
                signing_info.signature_verified,                # signature_verified
                datetime.strptime(signing_info.date_signed, "%Y-%m-%d %H:%M:%S") if signing_info.date_signed else None,  # date_signed
                signing_info.number_of_certificates,            # number_of_certificates
                signing_info.spcSpOpusInfo.get('ProgramName', None) if signing_info.spcSpOpusInfo else None,  # program_name
                signing_info.spcSpOpusInfo.get('MoreInfo', None) if signing_info.spcSpOpusInfo else None,    # more_info
                current_time                                    # analysis_date
            ]]

            # Prepare data for redb_pe_certificates table
            certificates_data = []
            
            # Process signers
            if signing_info.signers:
                for signer in signing_info.signers:
                    certificates_data.append(self._prepare_certificate_data(signer, 'signer', current_time))
            
            # Process countersigners
            if signing_info.countersigners:
                for csigner in signing_info.countersigners:
                    certificates_data.append(self._prepare_certificate_data(csigner, 'countersigner', current_time))
            
            # Process x509 certificates
            if signing_info.x509_certificates:
                for cert in signing_info.x509_certificates:
                    certificates_data.append(self._prepare_certificate_data(cert, 'chain', current_time))
            
            return {
                'multi_table': True,
                'signatures': {
                    'table': 'redb_pe_signatures',
                    'data': signatures_data,
                    'column_names': [
                        'sha256', 'md5', 'sha1', 'signature_verified', 'date_signed',
                        'number_of_certificates', 'spcSpOpusInfo_program_name', 'spcSpOpusInfo_more_info', 'analysis_date'
                    ],
                    'column_type_names': [
                        'FixedString(64)', 'FixedString(32)', 'FixedString(40)',
                        'Boolean', 'DateTime64(3, \'UTC\')', 'UInt8',
                        'Nullable(String)', 'Nullable(String)', 'DateTime64(3, \'UTC\')'
                    ]
                },
                'certificates': {
                    'table': 'redb_pe_certificates',
                    'data': certificates_data,
                    'column_names': [
                        'sha256', 'md5', 'sha1', 'certificate_type', 'certificate_algorithm',
                        'certificate_issuer', 'certificate_subject', 'certificate_serial_number',
                        'certificate_thumbprint', 'certificate_valid_from', 'certificate_valid_to',
                        'analysis_date'
                    ],
                    'column_type_names': [
                        'FixedString(64)', 'FixedString(32)', 'FixedString(40)',
                        'Enum8(\'signer\' = 1, \'countersigner\' = 2, \'chain\' = 3)', 'LowCardinality(String)',
                        'LowCardinality(String)', 'LowCardinality(String)', 'String',
                        'String', 'DateTime64(3, \'UTC\')', 'DateTime64(3, \'UTC\')',
                        'DateTime64(3, \'UTC\')'
                    ]
                }
            }

    # def _prepare_certificate_data(self, cert, cert_type: str, analysis_time: datetime) -> List:
    #     """Helper method to prepare certificate data for ClickHouse"""
    #     self.log.debug(inspect.currentframe().f_code.co_name)
    #     prefix = "certificate_" if hasattr(cert, "certificate_algorithm") else "signer_"
        
    #     # Get serial number and clean it up - with strict handling
    #     serial_number = getattr(cert, f"{prefix}serial_number", "UNKNOWN")
    #     if not serial_number or not str(serial_number).strip():
    #         serial_number = "UNKNOWN"
    #     else:
    #         serial_number = serial_number.replace(':', '')
    #         if not serial_number.strip():  # Double check after cleaning
    #             serial_number = "UNKNOWN"
        
    #     return [
    #         self.sha256,                          # sha256
    #         self.md5,                             # md5
    #         self.sha1,                            # sha1
    #         cert_type,                            # certificate_type
    #         getattr(cert, f"{prefix}algorithm"),   # certificate_algorithm
    #         getattr(cert, f"{prefix}issuer"),     # certificate_issuer
    #         getattr(cert, f"{prefix}subject"),    # certificate_subject
    #         serial_number,                        # certificate_serial_number
    #         getattr(cert, f"{prefix}thumbprint"),  # certificate_thumbprint
    #         datetime.strptime(getattr(cert, f"{prefix}valid_from"), "%Y-%m-%d %H:%M:%S"),  # certificate_valid_from
    #         datetime.strptime(getattr(cert, f"{prefix}valid_to"), "%Y-%m-%d %H:%M:%S"),    # certificate_valid_to
    #         analysis_time                         # analysis_date
    #     ]

    def _prepare_certificate_data(self, cert, cert_type: str, analysis_time: datetime) -> List:
        """Helper method to prepare certificate data for ClickHouse"""
        self.log.debug(inspect.currentframe().f_code.co_name)
        prefix = "certificate_" if hasattr(cert, "certificate_algorithm") else "signer_"
        
        def clean_string(value, field_name):
            if value is None or (isinstance(value, str) and not value.strip()):
                self.log.warning(f"Empty or None value found for {field_name}")
                return "UNKNOWN"
            return str(value).strip()
        
        try:
            serial_number = getattr(cert, f"{prefix}serial_number", "UNKNOWN")
            serial_number = clean_string(serial_number, "serial_number")
            if serial_number != "UNKNOWN":
                serial_number = serial_number.replace(':', '')
        
            return [
                clean_string(self.sha256, "sha256"),
                clean_string(self.md5, "md5"),
                clean_string(self.sha1, "sha1"),
                cert_type,
                clean_string(getattr(cert, f"{prefix}algorithm"), "algorithm"),
                clean_string(getattr(cert, f"{prefix}issuer"), "issuer"),
                clean_string(getattr(cert, f"{prefix}subject"), "subject"),
                serial_number,
                clean_string(getattr(cert, f"{prefix}thumbprint"), "thumbprint"),
                datetime.strptime(getattr(cert, f"{prefix}valid_from"), "%Y-%m-%d %H:%M:%S"),
                datetime.strptime(getattr(cert, f"{prefix}valid_to"), "%Y-%m-%d %H:%M:%S"),
                analysis_time
            ]
        except Exception as e:
            self.log.error(f"Error preparing certificate data: {str(e)}")
            self.log.error(f"Certificate data that caused error: {cert}")
            raise

    def get_clickhouse_table(self) -> str:
        """Required by PEExtractor abstract base class"""
        return "redb_pe_signatures"  # Return the primary table

    def get_clickhouse_tables(self) -> Dict[str, str]:
        """Additional method for multi-table support"""
        return {
            'signatures': 'redb_pe_signatures',
            'certificates': 'redb_pe_certificates'
        }