ScanPe

Collects metadata from PE files.

Source code in strelka/src/python/strelka/scanners/scan_pe.py

class ScanPe(strelka.Scanner):
    """Collects metadata from PE files."""

    def scan(self, data, file, options, expire_at):
        extract_overlay = options.get("extract_overlay", False)

        try:
            pe = pefile.PE(data=data)
            if not pe:
                self.flags.append("pe_load_error")
                return
        except pefile.PEFormatError:
            self.flags.append("pe_format_error")
            return
        except AttributeError:
            self.flags.append("pe_attribute_error")
            return

        if rich_dict := parse_rich(pe):
            if type(rich_dict) is not str:
                self.event["rich"] = rich_dict
            else:
                self.flags.append(rich_dict)

        if cert_dict := parse_certificates(data):
            if type(cert_dict) is not str:
                self.event["security"] = cert_dict
            else:
                self.flags.append(cert_dict)

        self.event["total"] = {
            "libraries": 0,
            "resources": 0,
            "sections": len(pe.sections),
            "symbols": 0,
        }
        self.event["summary"] = {}

        offset = pe.get_overlay_data_start_offset()

        if offset and len(data[offset:]) > 0:
            self.event["overlay"] = {"size": len(data[offset:]), "extracted": False}
            self.flags.append("overlay")

            if extract_overlay:
                # Send extracted file back to Strelka
                self.emit_file(data[offset:], name="pe_overlay")
                self.event["overlay"].update({"extracted": True})

        if hasattr(pe, "DIRECTORY_ENTRY_DEBUG"):
            for d in pe.DIRECTORY_ENTRY_DEBUG:
                try:
                    data = pe.get_data(d.struct.AddressOfRawData, d.struct.SizeOfData)
                    if data.find(b"RSDS") != -1 and len(data) > 24:
                        pdb = data[data.find(b"RSDS") :]
                        self.event["debug"] = {
                            "type": "rsds",
                            "guid": b"%s-%s-%s-%s"
                            % (
                                binascii.hexlify(pdb[4:8]),
                                binascii.hexlify(pdb[8:10]),
                                binascii.hexlify(pdb[10:12]),
                                binascii.hexlify(pdb[12:20]),
                            ),
                            "age": struct.unpack("<L", pdb[20:24])[0],
                            "pdb": pdb[24:].split(b"\x00")[0],
                        }
                    elif data.find(b"NB10") != -1 and len(data) > 16:
                        pdb = data[data.find(b"NB10") + 8 :]
                        self.event["debug"] = {
                            "type": "nb10",
                            "created": struct.unpack("<L", pdb[0:4])[0],
                            "age": struct.unpack("<L", pdb[4:8])[0],
                            "pdb": pdb[8:].split(b"\x00")[0],
                        }
                except pefile.PEFormatError:
                    self.flags.append("corrupt_debug_header")

        self.event["file_info"] = {
            "fixed": {},
            "string": [],
            "var": {},
        }

        # https://github.com/erocarrera/pefile/blob/master/pefile.py#L3553
        if hasattr(pe, "FileInfo"):
            if pe.FileInfo:
                fi = pe.FileInfo[0]  # contains a single element
                for i in fi:
                    if i.Key == b"StringFileInfo":
                        for st in i.StringTable:
                            for k, v in st.entries.items():
                                if k.decode() in COMMON_FILE_INFO_NAMES:
                                    self.event["file_info"][
                                        COMMON_FILE_INFO_NAMES[k.decode()]
                                    ] = v.decode()
                                else:
                                    self.event["file_info"]["string"].append(
                                        {
                                            "name": k.decode(),
                                            "value": v.decode(),
                                        }
                                    )
                    elif i.Key == b"VarFileInfo":
                        for v in i.Var:
                            if translation := v.entry.get(b"Translation"):
                                (lang, char) = translation.split()
                                self.event["file_info"]["var"] = {
                                    "language": VAR_FILE_INFO_LANGS.get(int(lang, 16)),
                                    "character_set": VAR_FILE_INFO_CHARS.get(
                                        int(char, 16)
                                    ),
                                }

        if hasattr(pe, "VS_FIXEDFILEINFO"):
            vs_ffi = pe.VS_FIXEDFILEINFO[0]  # contains a single element
            self.event["file_info"]["fixed"] = {
                "flags": [],
                "operating_systems": [],
                "type": {
                    "primary": FIXED_FILE_INFO_TYPE.get(vs_ffi.FileType),
                    "sub": FIXED_FILE_INFO_SUBTYPE.get(
                        (vs_ffi.FileType, vs_ffi.FileSubtype), ""
                    ),
                },
            }

            # http://www.jasinskionline.com/windowsapi/ref/v/vs_fixedfileinfo.html
            ff_flags = vs_ffi.FileFlagsMask & vs_ffi.FileFlags
            for f in FIXED_FILE_INFO_FLAGS:
                if ff_flags & f:
                    self.event["file_info"]["fixed"]["flags"].append(
                        FIXED_FILE_INFO_FLAGS[f]
                    )
            for o in FIXED_FILE_INFO_OS:
                if vs_ffi.FileOS & o:
                    self.event["file_info"]["fixed"]["operating_systems"].append(
                        FIXED_FILE_INFO_OS[o]
                    )

        self.event["header"] = {
            "machine": {
                "id": pe.FILE_HEADER.Machine,
                "type": pefile.MACHINE_TYPE.get(pe.FILE_HEADER.Machine, "").replace(
                    "IMAGE_FILE_MACHINE_", ""
                ),
            },
            "magic": {
                "dos": MAGIC_DOS.get(pe.DOS_HEADER.e_magic, ""),
                "image": MAGIC_IMAGE.get(pe.OPTIONAL_HEADER.Magic, ""),
            },
            "subsystem": pefile.SUBSYSTEM_TYPE.get(
                pe.OPTIONAL_HEADER.Subsystem, ""
            ).replace("IMAGE_SUBSYSTEM_", ""),
        }

        self.event["base_of_code"] = pe.OPTIONAL_HEADER.BaseOfCode
        self.event["address_of_entry_point"] = pe.OPTIONAL_HEADER.AddressOfEntryPoint
        self.event["image_base"] = pe.OPTIONAL_HEADER.ImageBase
        self.event["size_of_code"] = pe.OPTIONAL_HEADER.SizeOfCode
        self.event["size_of_initialized_data"] = (
            pe.OPTIONAL_HEADER.SizeOfInitializedData
        )
        self.event["size_of_headers"] = pe.OPTIONAL_HEADER.SizeOfHeaders
        self.event["size_of_heap_reserve"] = pe.OPTIONAL_HEADER.SizeOfHeapReserve
        self.event["size_of_image"] = pe.OPTIONAL_HEADER.SizeOfImage
        self.event["size_of_stack_commit"] = pe.OPTIONAL_HEADER.SizeOfStackCommit
        self.event["size_of_stack_reserve"] = pe.OPTIONAL_HEADER.SizeOfStackReserve
        self.event["size_of_heap_commit"] = pe.OPTIONAL_HEADER.SizeOfHeapCommit
        self.event["size_of_uninitialized_data"] = (
            pe.OPTIONAL_HEADER.SizeOfUninitializedData
        )
        self.event["file_alignment"] = pe.OPTIONAL_HEADER.FileAlignment
        self.event["section_alignment"] = pe.OPTIONAL_HEADER.SectionAlignment
        self.event["checksum"] = pe.OPTIONAL_HEADER.CheckSum

        self.event["major_image_version"] = pe.OPTIONAL_HEADER.MajorImageVersion
        self.event["minor_image_version"] = pe.OPTIONAL_HEADER.MinorImageVersion
        self.event["major_linker_version"] = pe.OPTIONAL_HEADER.MajorLinkerVersion
        self.event["minor_linker_version"] = pe.OPTIONAL_HEADER.MinorLinkerVersion
        self.event["major_operating_system_version"] = (
            pe.OPTIONAL_HEADER.MajorOperatingSystemVersion
        )
        self.event["minor_operating_system_version"] = (
            pe.OPTIONAL_HEADER.MinorOperatingSystemVersion
        )
        self.event["major_subsystem_version"] = pe.OPTIONAL_HEADER.MajorSubsystemVersion
        self.event["minor_subsystem_version"] = pe.OPTIONAL_HEADER.MinorSubsystemVersion
        self.event["image_version"] = float(
            f"{pe.OPTIONAL_HEADER.MajorImageVersion}.{pe.OPTIONAL_HEADER.MinorImageVersion}"
        )
        self.event["linker_version"] = float(
            f"{pe.OPTIONAL_HEADER.MajorLinkerVersion}.{pe.OPTIONAL_HEADER.MinorLinkerVersion}"
        )
        self.event["operating_system_version"] = float(
            f"{pe.OPTIONAL_HEADER.MajorOperatingSystemVersion}.{pe.OPTIONAL_HEADER.MinorOperatingSystemVersion}"
        )
        self.event["subsystem_version"] = float(
            f"{pe.OPTIONAL_HEADER.MajorSubsystemVersion}.{pe.OPTIONAL_HEADER.MinorSubsystemVersion}"
        )

        try:
            self.event["compile_time"] = datetime.datetime.utcfromtimestamp(
                pe.FILE_HEADER.TimeDateStamp
            ).isoformat()
        except OverflowError:
            self.flags.append("invalid compile time caused an overflow error")

        if hasattr(pe.OPTIONAL_HEADER, "BaseOfData"):
            self.event["base_of_data"] = pe.OPTIONAL_HEADER.BaseOfData

        dll_characteristics = []
        for o in CHARACTERISTICS_DLL:
            if pe.OPTIONAL_HEADER.DllCharacteristics & o:
                dll_characteristics.append(CHARACTERISTICS_DLL[o])

        if dll_characteristics:
            self.event["dll_characteristics"] = dll_characteristics

        image_characteristics = []
        for o in CHARACTERISTICS_IMAGE:
            if pe.FILE_HEADER.Characteristics & o:
                image_characteristics.append(CHARACTERISTICS_IMAGE[o])

        if image_characteristics:
            self.event["image_characteristics"] = image_characteristics

        self.event["resources"] = []
        if hasattr(pe, "DIRECTORY_ENTRY_RESOURCE"):
            resource_md5_set = set()
            resource_sha1_set = set()
            resource_sha256_set = set()

            for res0 in pe.DIRECTORY_ENTRY_RESOURCE.entries:
                if hasattr(res0, "directory"):
                    for res1 in res0.directory.entries:
                        if hasattr(res1, "directory"):
                            for res2 in res1.directory.entries:
                                lang = res2.data.lang
                                sub = res2.data.sublang
                                sub = pefile.get_sublang_name_for_lang(lang, sub)
                                try:
                                    data = pe.get_data(
                                        res2.data.struct.OffsetToData,
                                        res2.data.struct.Size,
                                    )
                                except pefile.PEFormatError:
                                    continue
                                resource_md5 = hashlib.md5(data).hexdigest()
                                resource_sha1 = hashlib.sha1(data).hexdigest()
                                resource_sha256 = hashlib.sha256(data).hexdigest()

                                resource_md5_set.add(resource_md5)
                                resource_sha1_set.add(resource_sha1)
                                resource_sha256_set.add(resource_sha256)

                                resource_dict = {
                                    "id": res1.id,
                                    "language": {"sub": sub.replace("SUBLANG_", "")},
                                    "type": pefile.RESOURCE_TYPE.get(
                                        res0.id, ""
                                    ).replace("RT_", ""),
                                    "md5": resource_md5,
                                    "sha1": resource_sha1,
                                    "sha256": resource_sha256,
                                }

                                if lang in pefile.LANG:
                                    resource_dict["language"]["primary"] = pefile.LANG[
                                        lang
                                    ].replace("LANG_", "")

                                if res1.name:
                                    resource_dict["name"] = str(res1.name)

                                self.event["resources"].append(resource_dict)

                        # TODO: Add optional resource extraction

            self.event["summary"]["resource_md5"] = list(resource_md5_set)
            self.event["summary"]["resource_sha1"] = list(resource_sha1_set)
            self.event["summary"]["resource_sha256"] = list(resource_sha256_set)

        self.event["total"]["resources"] = len(self.event["resources"])

        self.event["sections"] = []
        section_md5_set = set()
        section_sha1_set = set()
        section_sha256_set = set()

        for sec in pe.sections:
            try:
                name = sec.Name.rstrip(b"\x00").decode()
                section_md5 = sec.get_hash_md5()
                section_sha1 = sec.get_hash_sha1()
                section_sha256 = sec.get_hash_sha256()

                section_md5_set.add(section_md5)
                section_sha1_set.add(section_sha1)
                section_sha256_set.add(section_sha256)

                row = {
                    "address": {
                        "physical": sec.Misc_PhysicalAddress,
                        "virtual": sec.VirtualAddress,
                    },
                    "characteristics": [],
                    "entropy": sec.get_entropy(),
                    "name": name,
                    "size": sec.SizeOfRawData,
                    "md5": section_md5,
                    "sha1": section_sha1,
                    "sha256": section_sha256,
                }
                for o in CHARACTERISTICS_SECTION:
                    if sec.Characteristics & o:
                        row["characteristics"].append(CHARACTERISTICS_SECTION[o])

                # TODO: Add optional resource extraction

                self.event["sections"].append(row)
                self.event["summary"]["section_md5"] = list(section_md5_set)
                self.event["summary"]["section_sha1"] = list(section_sha1_set)
                self.event["summary"]["section_sha256"] = list(section_sha256_set)
            except strelka.ScannerTimeout:
                raise
            except Exception as e:
                self.flags.append(f"exception thrown when parsing section's {e}")

        self.event["symbols"] = {
            "exported": [],
            "imported": [],
            "libraries": [],
            "table": [],
        }

        if hasattr(pe, "DIRECTORY_ENTRY_IMPORT"):
            self.event["imphash"] = pe.get_imphash()

            for imp in pe.DIRECTORY_ENTRY_IMPORT:
                lib = imp.dll.decode()
                if lib not in self.event["symbols"]["libraries"]:
                    self.event["symbols"]["libraries"].append(lib)

                row = {
                    "library": lib,
                    "symbols": [],
                    "type": "import",
                }
                for e in imp.imports:
                    if not e.name:
                        name = f"ord{e.ordinal}"
                    else:
                        name = e.name.decode()
                    self.event["symbols"]["imported"].append(name)
                    row["symbols"].append(name)
                self.event["symbols"]["table"].append(row)

        if hasattr(pe, "DIRECTORY_ENTRY_EXPORT"):
            self.event["dll_name"] = pe.DIRECTORY_ENTRY_EXPORT.name
            for exp in pe.DIRECTORY_ENTRY_EXPORT.symbols:
                if not exp.name:
                    name = f"ord{exp.ordinal}"
                else:
                    name = exp.name
                self.event["symbols"]["exported"].append(name)
                self.event["symbols"]["table"].append(
                    {
                        "address": exp.address,
                        "symbol": name,
                        "type": "export",
                    }
                )

        self.event["total"]["libraries"] = len(self.event["symbols"]["libraries"])
        self.event["total"]["symbols"] = len(self.event["symbols"]["table"])

Features

The features of this scanner are detailed below. These features represent the capabilities and the type of analysis the scanner can perform. This may include support for Indicators of Compromise (IOC), the ability to emit files for further analysis, and the presence of extended documentation for complex analysis techniques.

Feature	Support
IOC Support
Emit Files
Extended Docs
Malware Scanner
Image Thumbnails

Tastes

Strelka's file distribution system assigns scanners to files based on 'flavors' and 'tastes'. Flavors describe the type of file, typically determined by MIME types from libmagic, matches from YARA rules, or characteristics of parent files. Tastes are the criteria used within Strelka to determine which scanners are applied to which files, with positive and negative tastes defining files to be included or excluded respectively.

Source Filetype	Include / Exclude
application/x-dosexec
mz_file

Scanner Fields

This section provides a list of fields that are extracted from the files processed by this scanner. These fields include the data elements that the scanner extracts from each file, representing the analytical results produced by the scanner. If the test file is missing or cannot be parsed, this section will not contain any data.

Field Name	Field Type
address_of_entry_point	int
base_of_code	int
checksum	int
compile_time	str
debug	dict
debug.age	int
debug.guid	bytes
debug.pdb	bytes
debug.type	str
dll_characteristics	str
elapsed	str
file_alignment	int
file_info	dict
file_info.assembly_version	str
file_info.comments	str
file_info.company_name	str
file_info.file_description	str
file_info.file_version	str
file_info.fixed	dict
file_info.fixed.flags	list
file_info.fixed.operating_systems	list
file_info.fixed.type	dict
file_info.fixed.type.primary	str
file_info.fixed.type.sub	str
file_info.internal_name	str
file_info.legal_copyright	str
file_info.legal_trademarks	str
file_info.original_filename	str
file_info.product_name	str
file_info.product_version	str
file_info.string	list
file_info.var	dict
file_info.var.character_set	str
file_info.var.language	NoneType
flags	list
header	dict
header.machine	dict
header.machine.id	int
header.machine.type	str
header.magic	dict
header.magic.dos	str
header.magic.image	str
header.subsystem	str
image_base	int
image_characteristics	str
image_version	float
linker_version	float
major_image_version	int
major_linker_version	int
major_operating_system_version	int
major_subsystem_version	int
minor_image_version	int
minor_linker_version	int
minor_operating_system_version	int
minor_subsystem_version	int
operating_system_version	float
overlay	dict
overlay.extracted	bool
overlay.size	int
resources	str
section_alignment	int
sections	str
size_of_code	int
size_of_headers	int
size_of_heap_commit	int
size_of_heap_reserve	int
size_of_image	int
size_of_initialized_data	int
size_of_stack_commit	int
size_of_stack_reserve	int
size_of_uninitialized_data	int
subsystem_version	float
summary	dict
summary.resource_md5	str
summary.resource_sha1	str
summary.resource_sha256	str
summary.section_md5	str
summary.section_sha1	str
summary.section_sha256	str
symbols	dict
symbols.exported	list
symbols.imported	list
symbols.libraries	list
symbols.table	list
total	dict
total.libraries	int
total.resources	int
total.sections	int
total.symbols	int

Sample Event

Below is a sample event generated by this scanner, demonstrating the kind of output that can be expected when it processes a file. This sample is derived from a mock scan event configured in the scanner's test file. If no test file is available, this section will not display a sample event.

    test_scan_event = {
        "elapsed": 0.001,
        "flags": ["no_certs_found"],
        "total": {"libraries": 0, "resources": 2, "sections": 2, "symbols": 0},
        "summary": {
            "resource_md5": unordered(
                [
                    "f4741884351459aa7733725b88e693af",
                    "b7db84991f23a680df8e95af8946f9c9",
                ]
            ),
            "resource_sha1": unordered(
                [
                    "5371904ee7671fb0b066d9323eda553269f344f9",
                    "cac699787884fb993ced8d7dc47b7c522c7bc734",
                ]
            ),
            "resource_sha256": unordered(
                [
                    "539dc26a14b6277e87348594ab7d6e932d16aabb18612d77f29fe421a9f1d46a",
                    "d8df3d0358a91b3ef97c4d472b34a60f7cf9ee7f1a6f37058fc3d1af3a156a36",
                ]
            ),
            "section_md5": unordered(
                [
                    "c3eafa2cd34f98a226e31b8ea3fea400",
                    "cc14da7fb94ef9b27a926fe95b86b44f",
                ]
            ),
            "section_sha1": unordered(
                [
                    "3d584b265a558dc22fa6dfa9991ae7eafee5c1a4",
                    "00104b432a8e7246695843e4f2d7cf2582efa3e6",
                ]
            ),
            "section_sha256": unordered(
                [
                    "86d9755b2ba9d8ffd765621f09844dd62d0b082fdc4aafa63b3b3f3ae25d9c77",
                    "bb31a5224e9f78905909655d9c80ba7d63f03910e4f22b296d6b7865e2a477c3",
                ]
            ),
        },
        "debug": {
            "type": "rsds",
            "guid": b"a66307d0-9b84-b944-bf030bff2d7d1e4a",
            "age": 1,
            "pdb": b"C:\\Users\\tmcguff\\source\\repos\\HelloWorld\\HelloWorld\\obj\\x64\\Release\\HelloWorld.pdb",
        },
        "file_info": {
            "fixed": {
                "flags": [],
                "operating_systems": ["WINDOWS32"],
                "type": {"primary": "APP", "sub": ""},
            },
            "string": [],
            "var": {"language": None, "character_set": "Unicode"},
            "comments": "",
            "company_name": ".",
            "file_description": "HelloWorld",
            "file_version": "1.0.0.0",
            "internal_name": "HelloWorld.exe",
            "legal_copyright": "Copyright © . 2020",
            "legal_trademarks": "",
            "original_filename": "HelloWorld.exe",
            "product_name": "HelloWorld",
            "product_version": "1.0.0.0",
            "assembly_version": "1.0.0.0",
        },
        "header": {
            "machine": {"id": 34404, "type": "AMD64"},
            "magic": {"dos": "DOS", "image": "64_BIT"},
            "subsystem": "WINDOWS_CUI",
        },
        "base_of_code": 8192,
        "address_of_entry_point": 0,
        "image_base": 5368709120,
        "size_of_code": 2048,
        "size_of_initialized_data": 1536,
        "size_of_headers": 512,
        "size_of_heap_reserve": 1048576,
        "size_of_image": 24576,
        "size_of_stack_commit": 16384,
        "size_of_stack_reserve": 4194304,
        "size_of_heap_commit": 8192,
        "size_of_uninitialized_data": 0,
        "file_alignment": 512,
        "section_alignment": 8192,
        "checksum": 0,
        "major_image_version": 0,
        "minor_image_version": 0,
        "major_linker_version": 48,
        "minor_linker_version": 0,
        "major_operating_system_version": 4,
        "minor_operating_system_version": 0,
        "major_subsystem_version": 4,
        "minor_subsystem_version": 0,
        "image_version": 0.0,
        "linker_version": 48.0,
        "operating_system_version": 4.0,
        "subsystem_version": 4.0,
        "compile_time": "2104-07-18T17:22:04",
        "dll_characteristics": unordered(
            [
                "DYNAMIC_BASE",
                "NX_COMPAT",
                "NO_SEH",
                "TERMINAL_SERVER_AWARE",
            ]
        ),
        "image_characteristics": unordered(["EXECUTABLE_IMAGE", "LARGE_ADDRESS_AWARE"]),
        "resources": unordered(
            [
                {
                    "id": 1,
                    "language": {"sub": "NEUTRAL", "primary": "NEUTRAL"},
                    "type": "VERSION",
                    "md5": "f4741884351459aa7733725b88e693af",
                    "sha1": "5371904ee7671fb0b066d9323eda553269f344f9",
                    "sha256": "d8df3d0358a91b3ef97c4d472b34a60f7cf9ee7f1a6f37058fc3d1af3a156a36",
                },
                {
                    "id": 1,
                    "language": {"sub": "NEUTRAL", "primary": "NEUTRAL"},
                    "type": "MANIFEST",
                    "md5": "b7db84991f23a680df8e95af8946f9c9",
                    "sha1": "cac699787884fb993ced8d7dc47b7c522c7bc734",
                    "sha256": "539dc26a14b6277e87348594ab7d6e932d16aabb18612d77f29fe421a9f1d46a",
                },
            ]
        ),
        "sections": unordered(
            [
                {
                    "address": {"physical": 1743, "virtual": 8192},
                    "characteristics": ["CNT_CODE", "MEM_EXECUTE", "MEM_READ"],
                    "entropy": 4.621214196319175,
                    "name": ".text",
                    "size": 2048,
                    "md5": "cc14da7fb94ef9b27a926fe95b86b44f",
                    "sha1": "3d584b265a558dc22fa6dfa9991ae7eafee5c1a4",
                    "sha256": "bb31a5224e9f78905909655d9c80ba7d63f03910e4f22b296d6b7865e2a477c3",
                },
                {
                    "address": {"physical": 1472, "virtual": 16384},
                    "characteristics": ["CNT_INITIALIZED_DATA", "MEM_READ"],
                    "entropy": 4.09070377434219,
                    "name": ".rsrc",
                    "size": 1536,
                    "md5": "c3eafa2cd34f98a226e31b8ea3fea400",
                    "sha1": "00104b432a8e7246695843e4f2d7cf2582efa3e6",
                    "sha256": "86d9755b2ba9d8ffd765621f09844dd62d0b082fdc4aafa63b3b3f3ae25d9c77",
                },
            ]
        ),
        "symbols": {"exported": [], "imported": [], "libraries": [], "table": []},
    }