Xavier Granier

import inspect
import pefile

from hashlib import md5, sha1, sha256

from asn1crypto import cms, pem, x509, core

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


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,
        index_prefix=None,
        elastic_index=None,
        known_benign=False,
        known_malicious=False,
        pe=None,
    ):
        super().__init__(
            filepath,
            log,
            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)

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

    @staticmethod
    def _extract_certificate_info(cert_data):
        # self.log.debug(inspect.currentframe().f_code.co_name)

        certificate = x509.Certificate.load(cert_data)
        certificate_serial_number = format(certificate.serial_number, "x").upper()
        grouped_serial_number = ":".join(
            [
                certificate_serial_number[i : i + 2]
                for i in range(0, len(certificate_serial_number), 2)
            ]
        )
        return PECertificate(
            _id=format(certificate.serial_number, "x").upper(),
            certificate_serial_number=grouped_serial_number,
            certificate_issuer=certificate.issuer.human_friendly,
            certificate_subject=certificate.subject.human_friendly,
            certificate_valid_from=certificate["tbs_certificate"]["validity"][
                "not_before"
            ].native.strftime("%Y-%m-%d %H:%M:%S"),
            certificate_valid_to=certificate["tbs_certificate"]["validity"][
                "not_after"
            ].native.strftime("%Y-%m-%d %H:%M:%S"),
            certificate_thumbprint=sha1(cert_data).hexdigest().upper(),
            certificate_md5=md5(cert_data).hexdigest().upper(),
            certificate_sha256=sha256(cert_data).hexdigest().upper(),
        )

    @staticmethod
    def _parse_microsoft_time_stamp(ts_token):
        class TSTInfo(core.Sequence):
            _fields = [
                ("version", core.Integer),
                ("policy_id", core.ObjectIdentifier),
                ("message_imprint", core.Sequence),
                ("serial_number", core.Integer),
                ("gen_time", core.GeneralizedTime),
                ("accuracy", core.Sequence, {"optional": True}),
                ("ordering", core.Boolean, {"optional": True}),
                ("nonce", core.Integer, {"optional": True}),
                ("tsa", core.Sequence, {"optional": True}),
                ("extensions", core.Sequence, {"optional": True}),
            ]

        # The content should be SignedData
        signed_data = ts_token["content"]

        # The actual timestamp should be in the encapContentInfo
        encap_content_info = signed_data["encap_content_info"]
        tst_info = TSTInfo.load(encap_content_info["content"])

        # Extract the genTime field from TSTInfo, type(genTime): <class 'datetime.datetime'>
        gen_time = tst_info["gen_time"].native

        return gen_time

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

        for attr_set in [signer["signed_attrs"], signer["unsigned_attrs"]]:
            if attr_set:
                for attribute in attr_set:
                    attr_type = attribute["type"].dotted
                    if attr_type == "1.2.840.113549.1.9.5":  # signingTime
                        signing_time = attribute["values"][0].native
                        break
                    elif (
                        attr_type == "1.3.6.1.4.1.311.3.2.1"
                        or attr_type == "1.3.6.1.4.1.311.3.3.1"
                    ):  # microsoft_time_stamp_token
                        try:
                            ts_token = attribute["values"][0].native
                            signing_time = self._parse_microsoft_time_stamp(ts_token)
                            self.log.debug(
                                f"Signing Time from Microsoft Time Stamp Token: {signing_time}"
                            )
                        except Exception as e:
                            self.log.error(
                                f"Error parsing Microsoft Time Stamp Token: {e}"
                            )
                            self.log.error("Token structure:", attribute["values"][0])
                    elif attr_type == "1.2.840.113549.1.9.6":  # counterSignature
                        counter_sig_values = attribute["values"]
                        for counter_sig_value in counter_sig_values:
                            try:
                                counter_sig_info = cms.SignerInfo.load(
                                    counter_sig_value.native
                                )
                                cs_signed_attrs = counter_sig_info["signed_attrs"]
                                for cs_attr in cs_signed_attrs:
                                    cs_attr_type = cs_attr["type"].dotted
                                    if (
                                        cs_attr_type == "1.2.840.113549.1.9.5"
                                    ):  # signingTime
                                        signing_time = cs_attr["values"][0].native
                                        self.log.debug(
                                            "Signing Time (from countersignature):",
                                            signing_time,
                                        )
                                        break
                                if signing_time:
                                    break
                            except Exception as e:
                                self.log.error(
                                    f"Error processing countersignature: {e}"
                                )
                    if signing_time:
                        break
            if signing_time:
                break
        signer_serial_number = format(
            signer["sid"].native["serial_number"], "x"
        ).upper()
        grouped_serial_number = " ".join(
            [
                signer_serial_number[i : i + 2]
                for i in range(0, len(signer_serial_number), 2)
            ]
        )
        return PESigner(
            # this is to avoid duplicate _id with the Signer certificate used to sign as they are in the same ES index
            _id=format(signer["sid"].native["serial_number"], "x").upper()
            + signer["digest_algorithm"]["algorithm"].native,
            signer_serial_number=grouped_serial_number,
            signer_digest_algorithm=signer["digest_algorithm"]["algorithm"].native,
            signer_signature_algorithm=signer["signature_algorithm"][
                "algorithm"
            ].native,
            signing_time=(
                signing_time.strftime("%Y-%m-%d %H:%M:%S") if signing_time else None
            ),
            # "countersigners": extract_countersigners(signer), # TODO: Implement this
        )

    def _extract_signature_info(self):
        self.log.debug(inspect.currentframe().f_code.co_name)
        address = self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[
            pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_SECURITY"]
        ].VirtualAddress
        size = self.pe.OPTIONAL_HEADER.DATA_DIRECTORY[
            pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_SECURITY"]
        ].Size

        addr_8 = address + 8
        signature_data = bytes(
            self.pe.write()[addr_8 : addr_8 + size]  # noqa E203
        )  # Ensure this is a bytes object
        if pem.detect(signature_data):
            signature_data = pem.unarmor(signature_data)

        content_info = cms.ContentInfo.load(signature_data)
        signed_data = content_info["content"]

        certificates = signed_data["certificates"]
        signers = signed_data["signer_infos"]

        certificates_list = []
        for cert in certificates:
            cert_info = self._extract_certificate_info(cert.chosen.dump())
            certificates_list.append(cert_info)

        signer_info_list = []
        for signer in signers:
            signer_info = self._extract_signer_info(signer)
            signer_info_list.append(signer_info)

        return certificates_list, signer_info_list

    def extract(self):
        try:
            self.log.debug(inspect.currentframe().f_code.co_name)
            if self._is_signed():
                certificates_list, signers_list = self._extract_signature_info()
                # self.export_to_elastic([PECodeSigning(certificates_list, signers_list)])
                self.export_to_elastic(certificates_list, Tag.PE_CERTIFICATE.value)
                self.export_to_elastic(signers_list, Tag.PE_SIGNER.value)
            return True
        except Exception as e:
            self.log.error(f"Error extracting PE Signature: {e}")
            return None