Add .NET memory model

malduck/__init__.py

@@ -47,6 +47,7 @@
     cuckoomem,
     idamem,
     procmem,
+    procmemdnpe,
     procmemelf,
     procmempe,
 )
@@ -166,6 +167,7 @@
     # procmem
     "procmem",
     "procmempe",
+    "procmemdnpe",
     "procmemelf",
     "cuckoomem",
     "idamem",

malduck/dnpe.py

@@ -0,0 +1,85 @@
+from typing import Any, Iterator, List, Optional, Union
+
+import dnfile
+
+from .pe import PE, MemoryPEData
+from .procmem import ProcessMemory
+
+__all__ = ["dnpe", "DnPE", "MemoryDnPEData"]
+
+
+class MemoryDnPEData(MemoryPEData):
+    def __init__(self, memory: ProcessMemory, fast_load: bool) -> None:
+        self.memory = memory
+        # Preload headers
+        self.pe = dnfile.dnPE(data=self, fast_load=True)
+        if not fast_load:
+            self.pe.full_load()
+
+
+class DnPE(PE):
+    def __init__(
+        self, data: Union[ProcessMemory, bytes], fast_load: bool = False
+    ) -> None:
+
+        if isinstance(data, ProcessMemory):
+            self.pe = MemoryDnPEData(data, fast_load).pe
+        else:
+            self.pe = dnfile.dnPE(data=data, fast_load=fast_load)
+
+    @property
+    def dn_metadata(self) -> Optional[dnfile.stream.MetaDataTables]:
+        return self.pe.net.metadata
+
+    @property
+    def dn_strings(self) -> Optional[dnfile.stream.StringsHeap]:
+        return self.pe.net.strings
+
+    @property
+    def dn_user_strings(self) -> Optional[dnfile.stream.UserStringHeap]:
+        return self.pe.net.user_strings
+
+    @property
+    def dn_guid(self) -> Optional[dnfile.stream.GuidHeap]:
+        return self.pe.net.guids
+
+    @property
+    def dn_mdtables(self) -> Optional[dnfile.stream.MetaDataTables]:
+        return self.pe.net.mdtables
+
+    @property
+    def dn_resources(self) -> List:
+        return self.pe.net.resources
+
+    @property
+    def dn_flags(self) -> Any:
+        return self.pe.net.flags
+
+    def dn_user_string(
+        self, index: int, encoding="utf-16"
+    ) -> Optional[dnfile.stream.UserString]:
+        if not self.dn_user_strings or self.dn_user_strings.sizeof() == 0:
+            return None
+
+        try:
+            us_string = self.dn_user_strings.get_us(index, encoding=encoding)
+        except UnicodeDecodeError:
+            return None
+
+        return us_string
+
+    def dn_iterate_resources(self) -> Iterator:
+        for resource in self.dn_resources:
+            if isinstance(resource.data, bytes):
+                yield resource
+
+            elif isinstance(resource.data, dnfile.resource.ResourceSet):
+                if not resource.data.entries:
+                    continue
+
+                for entry in resource.data.entries:
+                    if entry.data:
+                        yield entry.data
+
+
+dnpe = DnPE

malduck/procmem/__init__.py

@@ -1,6 +1,7 @@
 from .cuckoomem import CuckooProcessMemory, cuckoomem
 from .idamem import IDAProcessMemory, idamem
 from .procmem import MemoryBuffer, ProcessMemory, procmem
+from .procmemdnpe import ProcessMemoryDnPE, procmemdnpe
 from .procmemelf import ProcessMemoryELF, procmemelf
 from .procmempe import ProcessMemoryPE, procmempe
 from .region import (
@@ -19,6 +20,8 @@
     "procmem",
     "ProcessMemoryPE",
     "procmempe",
+    "ProcessMemoryDnPE",
+    "procmemdnpe",
     "MemoryBuffer",
     "ProcessMemoryELF",
     "procmemelf",

malduck/procmem/procmemdnpe.py

@@ -0,0 +1,62 @@
+from typing import List, Optional
+
+from ..dnpe import DnPE
+from .binmem import ProcessMemoryBuffer
+from .procmempe import ProcessMemoryPE
+from .region import Region
+
+__all__ = ["ProcessMemoryDnPE", "procmemdnpe"]
+
+
+class ProcessMemoryDnPE(ProcessMemoryPE):
+
+    __magic__ = b"MZ"
+
+    def __init__(
+        self,
+        buf: ProcessMemoryBuffer,
+        base: int = 0,
+        regions: Optional[List[Region]] = None,
+        image: bool = False,
+        detect_image: bool = False,
+    ) -> None:
+        self._pe: Optional[DnPE] = None
+        super(ProcessMemoryPE, self).__init__(
+            buf, base=base, regions=regions, image=image, detect_image=detect_image
+        )
+
+    def _pe_direct_load(self, fast_load: bool = True) -> DnPE:
+        offset = self.v2p(self.imgbase)
+        if offset is None:
+            raise ValueError("imgbase out of regions")
+        # Expected m type: bytearray
+        m = bytearray(self.readp(offset))
+        pe = DnPE(data=m, fast_load=fast_load)
+        return pe
+
+    def is_valid(self) -> bool:
+        if self.readv(self.imgbase, 2) != self.__magic__:
+            return False
+        pe_offs = self.uint32v(self.imgbase + 0x3C)
+        if pe_offs is None:
+            return False
+        if self.readv(self.imgbase + pe_offs, 2) != b"PE":
+            return False
+        try:
+            dn = DnPE(self)
+            if not hasattr(dn, "net"):
+                return False
+
+            return True
+        except Exception:
+            return False
+
+    @property
+    def pe(self) -> DnPE:
+        """Related :class:`PE` object"""
+        if self._pe is None:
+            self._pe = DnPE(self)
+        return self._pe
+
+
+procmemdnpe = ProcessMemoryDnPE

requirements.txt

@@ -6,3 +6,4 @@ capstone>=4.0.1
 yara-python
 typing-extensions>=3.7.4.2
 cryptography>=3.1
+dnfile==0.11.0

tests/test_procmem.py

@@ -7,7 +7,7 @@
 import tempfile
 import pytest
 
-from malduck import procmem, procmempe, cuckoomem, pad, pe, insn, PAGE_READWRITE, enhex
+from malduck import procmem, procmempe, procmemdnpe, cuckoomem, pad, pe, insn, PAGE_READWRITE, enhex
 from malduck.procmem import Region
 
 
@@ -263,3 +263,12 @@ def test_procmempe_image_sections():
 
     with procmempe.from_file("tests/files/96emptysections.exe.bin") as p:
         assert p.readv(0x2000, 8) == b"\x68\x28\x20\x40\x00\xe8\x0e\x00"
+
+
+def test_procmemdnpe():
+    with procmemdnpe.from_file("tests/files/dn_hello.exe", image=True) as p:
+        assert p is not None
+        assert p.pe is not None
+        assert p.pe.dn_metadata.struct.Version == b'v4.0.30319\x00\x00'
+        assert p.pe.dn_metadata.struct.NumberOfStreams == len(p.pe.dn_metadata.streams)
+