Initial commit

Author: Schamper

dissect/executable/pe/__init__.py

@@ -0,0 +1,5 @@
+from dissect.executable.pe.pe import PE
+
+__all__ = [
+    "PE",
+]

dissect/executable/pe/c_pe.py

@@ -0,0 +1,41 @@
+from dissect.executable.pe.directory.base import DataDirectory
+from dissect.executable.pe.directory.basereloc import BaseRelocationDirectory
+from dissect.executable.pe.directory.bound_import import BoundImportDirectory
+from dissect.executable.pe.directory.com_descriptor import ComDescriptorDirectory
+from dissect.executable.pe.directory.debug import DebugDirectory
+from dissect.executable.pe.directory.delay_import import DelayImportDirectory
+from dissect.executable.pe.directory.exception import ExceptionDirectory
+from dissect.executable.pe.directory.export import ExportDirectory
+from dissect.executable.pe.directory.iat import IatDirectory
+from dissect.executable.pe.directory.imports import ImportDirectory, ImportFunction, ImportModule
+from dissect.executable.pe.directory.load_config import LoadConfigDirectory
+from dissect.executable.pe.directory.resource import (
+    ResourceDataEntry,
+    ResourceDirectory,
+    ResourceDirectoryEntry,
+    ResourceEntry,
+)
+from dissect.executable.pe.directory.security import SecurityDirectory
+from dissect.executable.pe.directory.tls import TlsDirectory
+
+__all__ = [
+    "BaseRelocationDirectory",
+    "BoundImportDirectory",
+    "ComDescriptorDirectory",
+    "DataDirectory",
+    "DebugDirectory",
+    "DelayImportDirectory",
+    "ExceptionDirectory",
+    "ExportDirectory",
+    "IatDirectory",
+    "ImportDirectory",
+    "ImportFunction",
+    "ImportModule",
+    "LoadConfigDirectory",
+    "ResourceDataEntry",
+    "ResourceDirectory",
+    "ResourceDirectoryEntry",
+    "ResourceEntry",
+    "SecurityDirectory",
+    "TlsDirectory",
+]

dissect/executable/pe/c_pe.pyi

@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from dissect.executable.pe.pe import PE
+
+
+class DataDirectory:
+    """Base class for PE data directories."""
+
+    def __init__(self, pe: PE, address: int, size: int):
+        self.pe = pe
+        self.address = address
+        self.size = size
+
+    def __repr__(self) -> str:
+        return f"<{self.__class__.__name__} address={self.address:#x} size={self.size}>"

dissect/executable/pe/directory/__init__.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from functools import cached_property
+from typing import TYPE_CHECKING
+
+from dissect.executable.pe.c_pe import c_pe
+from dissect.executable.pe.directory.base import DataDirectory
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+
+class BaseRelocationDirectory(DataDirectory):
+    """The base relocation directory of a PE file."""
+
+    def __repr__(self) -> str:
+        return f"<BaseRelocationDirectory entries={len(self.entries)}>"
+
+    def __len__(self) -> int:
+        return len(self.entries)
+
+    def __iter__(self) -> Iterator[BaseRelocation]:
+        return iter(self.entries)
+
+    def __getitem__(self, idx: int) -> BaseRelocation:
+        return self.entries[idx]
+
+    @cached_property
+    def entries(self) -> list[BaseRelocation]:
+        """List of base relocation entries."""
+        result = []
+
+        offset = self.address
+        while offset < self.address + self.size:
+            self.pe.vfh.seek(offset)
+
+            block = c_pe._IMAGE_BASE_RELOCATION(self.pe.vfh)
+            if block.SizeOfBlock == 0:
+                break
+
+            page_rva = block.VirtualAddress
+
+            num_entries = (block.SizeOfBlock - len(c_pe._IMAGE_BASE_RELOCATION)) // len(c_pe.USHORT)
+            result.extend(
+                BaseRelocation(c_pe.IMAGE_REL_BASED(entry >> 12), page_rva + (entry & 0xFFF))
+                for entry in c_pe.USHORT[num_entries](self.pe.vfh)
+                if (entry >> 12) != 0  # Skip IMAGE_REL_BASED_ABSOLUTE (0)
+            )
+            offset += block.SizeOfBlock
+            offset += -offset & 3  # Align to 4 bytes
+
+        return result
+
+
+@dataclass
+class BaseRelocation:
+    """A single base relocation entry in the base relocation directory."""
+
+    type: c_pe.IMAGE_REL_BASED
+    rva: int
+
+    def __repr__(self) -> str:
+        return f"<BaseRelocation rva={self.rva:#x} type={self.type.name or self.type.value}>"

dissect/executable/pe/directory/base.py

@@ -0,0 +1,114 @@
+from __future__ import annotations
+
+from functools import cached_property
+from typing import TYPE_CHECKING
+
+from dissect.util.ts import from_unix
+
+from dissect.executable.pe.c_pe import c_pe
+from dissect.executable.pe.directory.base import DataDirectory
+
+if TYPE_CHECKING:
+    import datetime
+
+
+class BoundImportDirectory(DataDirectory):
+    """The bound import directory of a PE file."""
+
+    def __repr__(self) -> str:
+        return f"<BoundImportDirectory modules={len(self.modules)}>"
+
+    def __len__(self) -> int:
+        return len(self.modules)
+
+    def __getitem__(self, idx: str | int) -> BoundImportModule:
+        if isinstance(idx, int):
+            return self.modules[idx]
+        if isinstance(idx, str):
+            if idx not in self._by_name:
+                raise KeyError(f"Bound import module {idx!r} not found")
+            return self._by_name[idx]
+        raise TypeError(f"BoundImportDirectory indices must be str or int, not {type(idx).__name__}")
+
+    def __contains__(self, name: str) -> bool:
+        if isinstance(name, str):
+            return name in self._by_name
+        return False
+
+    @cached_property
+    def modules(self) -> list[BoundImportModule]:
+        """List of bound imported modules."""
+        result = []
+
+        self.pe.vfh.seek(self.address)
+        while self.pe.vfh.tell() < self.address + self.size:
+            descriptor = c_pe.IMAGE_BOUND_IMPORT_DESCRIPTOR(self.pe.vfh)
+            if not descriptor:
+                break
+
+            forwarders = []
+            for _ in range(descriptor.NumberOfModuleForwarderRefs):
+                forwarder = c_pe.IMAGE_BOUND_FORWARDER_REF(self.pe.vfh)
+                if not forwarder:
+                    break
+
+                forwarders.append(BoundImportForwardReference(self, forwarder))
+
+            result.append(BoundImportModule(self, descriptor, forwarders))
+
+        return result
+
+    @cached_property
+    def _by_name(self) -> dict[str, BoundImportModule]:
+        """A mapping of module names to their :class:`DelayImportModule`."""
+        return {module.name: module for module in self.modules}
+
+
+class BoundImportModule:
+    """A module bound imported by a PE file, containing its functions."""
+
+    def __init__(
+        self,
+        directory: BoundImportDirectory,
+        descriptor: c_pe.IMAGE_BOUND_IMPORT_DESCRIPTOR,
+        forwarders: list[BoundImportForwardReference],
+    ):
+        self.directory = directory
+        self.descriptor = descriptor
+        self.forwarders = forwarders
+
+    @property
+    def timestamp(self) -> datetime.datetime | None:
+        """The timestamp of this bound import module, or ``None`` if the PE file is compiled as reproducible."""
+        if self.directory.pe.is_reproducible():
+            return None
+        return from_unix(self.descriptor.TimeDateStamp)
+
+    @property
+    def name(self) -> str:
+        self.directory.pe.vfh.seek(self.directory.address + self.descriptor.OffsetModuleName)
+        return c_pe.CHAR[None](self.directory.pe.vfh).decode()
+
+
+class BoundImportForwardReference:
+    """A forward reference in a bound import module."""
+
+    def __init__(
+        self,
+        directory: BoundImportDirectory,
+        descriptor: c_pe.IMAGE_BOUND_FORWARDER_REF,
+    ):
+        self.directory = directory
+        self.descriptor = descriptor
+
+    @property
+    def timestamp(self) -> datetime.datetime | None:
+        """The timestamp of this bound import module, or ``None`` if the PE file is compiled as reproducible."""
+        if self.directory.pe.is_reproducible():
+            return None
+        return from_unix(self.descriptor.TimeDateStamp)
+
+    @property
+    def name(self) -> str:
+        self.directory.pe.vfh.seek(self.directory.address + self.descriptor.OffsetModuleName)
+        return c_pe.CHAR[None](self.directory.pe.vfh).decode()

dissect/executable/pe/directory/basereloc.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+from functools import cached_property
+from typing import BinaryIO
+
+from dissect.util.stream import RangeStream
+
+from dissect.executable.pe.c_pe import c_pe
+from dissect.executable.pe.directory.base import DataDirectory
+
+
+class ComDescriptorDirectory(DataDirectory):
+    """The COM descriptor directory of a PE file.
+
+    References:
+        - https://www.codeproject.com/Articles/12585/The-NET-File-Format
+    """
+
+    @cached_property
+    def descriptor(self) -> c_pe.IMAGE_COR20_HEADER:
+        """The CLR 2.0 header descriptor."""
+        self.pe.vfh.seek(self.address)
+        return c_pe.IMAGE_COR20_HEADER(self.pe.vfh)
+
+    @cached_property
+    def metadata(self) -> ComMetadata:
+        """The COM metadata directory."""
+        return ComMetadata(self.pe, self.descriptor.MetaData.VirtualAddress, self.descriptor.MetaData.Size)
+
+
+class ComMetadata(DataDirectory):
+    """The COM metadata directory of the COM descriptor."""
+
+    @cached_property
+    def metadata(self) -> c_pe.IMAGE_COR20_METADATA:
+        """The CLR 2.0 metadata descriptor."""
+        self.pe.vfh.seek(self.address)
+        return c_pe.IMAGE_COR20_METADATA(self.pe.vfh)
+
+    @property
+    def version(self) -> str:
+        """The version as defined in the metadata."""
+        return self.metadata.Version.decode().strip("\x00")
+
+    @cached_property
+    def streams(self) -> list[ComStream]:
+        """A list of streams defined in the metadata."""
+        result = []
+
+        offset = self.address + len(self.metadata)
+        for _ in range(self.metadata.NumberOfStreams):
+            self.pe.vfh.seek(offset)
+            header = c_pe.IMAGE_COR20_STREAM_HEADER(self.pe.vfh)
+
+            result.append(ComStream(self, header.Offset, header.Size, header.Name.decode()))
+
+            offset += len(header)
+            offset += -offset & 3  # Align to 4 bytes
+
+        return result
+
+
+class ComStream:
+    """A stream in the COM metadata."""
+
+    def __init__(self, metadata: ComMetadata, offset: int, size: int, name: str):
+        self.metadata = metadata
+        self.offset = offset
+        self.size = size
+        self.name = name
+
+    def __repr__(self) -> str:
+        return f"<ComStream offset={self.offset:#x} size={self.size:#x} name={self.name!r}>"
+
+    @property
+    def data(self) -> bytes:
+        """The data of the stream."""
+        self.metadata.pe.vfh.seek(self.metadata.address + self.offset)
+        return self.metadata.pe.vfh.read(self.size)
+
+    def open(self) -> BinaryIO:
+        """Open the stream for reading."""
+        return RangeStream(self.metadata.pe.vfh, self.metadata.address + self.offset, self.size)