Add minidump integration

cemu/emulator.py

@@ -1,18 +1,20 @@
 import collections
+import pathlib
+import udmp_parser
+import unicorn
+
 from enum import IntEnum, unique
 from multiprocessing import Lock
 from typing import Any, Callable, Optional
 
-import unicorn
-
 import cemu.const
 import cemu.core
 import cemu.os
 import cemu.utils
 from cemu.log import dbg, error, info, warn
 
-from .arch import is_x86, is_x86_32, x86
-from .memory import MemorySection
+from .arch import is_x86, is_x86_32, is_x86_64, x86
+from .memory import MemoryPermission, MemorySection
 
 
 @unique
@@ -628,3 +630,170 @@ def stop(self) -> None:
 
  self.vm.emu_stop()
  return
+
+ def allocate(self, size: int) -> int:
+ def aligned_allocation_size(size: int) -> int:
+ if size % ALLOCATION_ALIGNMENT == 0:
+ return size
+ return (size // ALLOCATION_ALIGNMENT) + ALLOCATION_ALIGNMENT
+
+ if not self.vm:
+ return -1
+
+ ALLOCATABLE_REGION_START = 0x1_0000
+ LAST_ALLOCATION = ALLOCATABLE_REGION_START
+ ALLOCATION_ALIGNMENT = 0x1_0000
+ MAX_ATTEMPT = 10
+
+ addr = LAST_ALLOCATION
+ sz = aligned_allocation_size(size)
+ for i in range(MAX_ATTEMPT):
+ try:
+ dbg(f"Trying to allocate({addr=:#x}, {sz=:d}), attempt {i}")
+ self.vm.mem_map(addr, sz, unicorn.UC_PROT_ALL)
+ LAST_ALLOCATION = addr + ALLOCATION_ALIGNMENT
+ return addr
+ except unicorn.unicorn.UcError as e:
+ print(f"UcError({str(e)}")
+ addr += ALLOCATION_ALIGNMENT
+ continue
+
+ raise Exception(f"Memory allocation: failed to malloc({size})")
+
+ def load_dumpfile(self, dmp_fpath: pathlib.Path) -> bool:
+ """Populate the VM with the execution context from a minidump file (if running under a Windows context) or a
+ coredump (if running under a Linux context)
+
+ Args:
+ dmp_fpath (pathlib.Path): _description_
+
+ Raises:
+ NotImplementedError
+
+ Returns:
+ bool: _description_
+ """
+ match cemu.core.context.os:
+ case cemu.os.Linux:
+ dbg(f"Parsing coredump {dmp_fpath}")
+ raise NotImplementedError("Linux coredump not implemented yet")
+
+ case cemu.os.Windows:
+ dbg(f"Parsing minidump {dmp_fpath}")
+ dmp = udmp_parser.UserDumpParser()
+ if not dmp.Parse(dmp_fpath):
+ error(f"Failed to parse minidump {dmp_fpath}")
+ return False
+
+ #
+ # Populate the memory
+ #
+ memory = dmp.Memory()
+ dbg(f"Mapping {len(memory)} memory sections")
+ self.sections.clear()
+
+ for _, section in memory.items():
+ content = bytes(
+ dmp.ReadMemory(section.BaseAddress, section.RegionSize)
+ )
+ section = MemorySection(
+ "",
+ section.BaseAddress,
+ section.RegionSize,
+ MemoryPermission.from_windows(section.Protect),
+ data_content=content,
+ )
+ self.sections.append(section)
+
+ #
+ # Populate the threads
+ #
+ self.threads = dmp.Threads()
+
+ tids = list(self.threads.keys())
+ if not self.switch_to_thread(tids[0]):
+ return False
+
+ case _:
+ raise ValueError("Unknown OS context")
+
+ return True
+
+ def switch_to_thread(self, tid: int) -> bool:
+ """Switch to the context of the thread whose TID is given as argument
+
+ Args:
+ tid (int): _description_
+
+ Raises:
+ IndexError: _description_
+ NotImplementedError: _description_
+ """
+ # https://wiki.osdev.org/SWAPGS
+ FSBase, GSBase, KernelGSBase = 0xC0000100, 0xC0000101, 0xC0000102 # noqa: F841
+
+ if not self.vm:
+ warn("VM must be initialized")
+ return False
+
+ threads = [t for t in self.threads if t.Tid == tid]
+ if not threads:
+ raise IndexError(f"No thread with TID={tid}")
+
+ assert len(threads) == 1, f"Multiple threads with TID={tid}, invalid"
+
+ thread = threads[0]
+
+ dbg(f"Trying to switch to thread context TID={thread.Tid}")
+
+ # TODO abstract thread switching through architecture
+ if not is_x86_64(cemu.core.context.architecture):
+ raise NotImplementedError(
+ "Thread switching only implemented for x64, currently"
+ )
+
+ try:
+ # Set GSBase to the TEB
+ self.vm.msr_write(GSBase, thread.Teb)
+ self.vm.reg_write(unicorn.x86_const.UC_X86_REG_CS, thread.Context.SegCs)
+ self.vm.reg_write(unicorn.x86_const.UC_X86_REG_DS, thread.Context.SegDs)
+ self.vm.reg_write(
+ unicorn.x86_const.UC_X86_REG_RIP, thread.Context.Rip
+ ) # pc
+ self.vm.reg_write(
+ unicorn.x86_const.UC_X86_REG_RSP, thread.Context.Rsp
+ ) # sp
+ # uc.reg_write(unicorn.x86_const.UC_X86_REG_GS, curthread.Context.SegGs)
+ # TODO other regs?
+ except Exception as e:
+ error(f"Exception: {str(e)}")
+ return False
+
+ return True
+
+ def invoke(self, start_address: int, end_address: int, args: list[Any]) -> bool:
+ if not self.vm:
+ warn("VM must be initialized")
+ return False
+
+ if not is_x86_64(cemu.core.context.architecture):
+ raise NotImplementedError(
+ "Thread switching only implemented for x64, currently"
+ )
+
+ # TODO abstract the calling convention through architecture
+ if len(args) >= 1:
+ self.vm.reg_write(unicorn.x86_const.UC_X86_REG_RCX, args[0])
+ if len(args) >= 2:
+ self.vm.reg_write(unicorn.x86_const.UC_X86_REG_RDX, args[1])
+ if len(args) >= 3:
+ self.vm.reg_write(unicorn.x86_const.UC_X86_REG_R8, args[2])
+ if len(args) >= 4:
+ self.vm.reg_write(unicorn.x86_const.UC_X86_REG_R8, args[3])
+
+ self.start_addr = start_address
+ self.end_addr = end_address
+
+ self.set(EmulatorState.RUNNING)
+
+ return True

cemu/memory.py

@@ -1,6 +1,6 @@
 import enum
 import pathlib
-from typing import Optional
+from typing import Optional, Union
 
 import unicorn
 
@@ -180,8 +180,9 @@ def __init__(
  name: str,
  addr: int,
  size: int,
- perm: str,
+ perm: Union[str, MemoryPermission],
  data_file: Optional[pathlib.Path] = None,
+ data_content: Optional[bytes] = None,
  ):
  if addr < 0 or addr >= 2**64:
  raise ValueError("address")
@@ -195,8 +196,16 @@ def __init__(
  self.name = name.strip().lower()
  self.address = addr
  self.size = size
- self.permission = MemoryPermission.from_string(perm)
+
+ if isinstance(perm, str):
+ self.permission = MemoryPermission.from_string(perm)
+ elif isinstance(perm, MemoryPermission):
+ self.permission = perm
+ else:
+ raise TypeError("Invalid type for `perm`")
+
  self.file_source = data_file if data_file and data_file.is_file() else None
+ self.__content: Optional[bytes] = data_content
  return
 
  @property
@@ -213,13 +222,25 @@ def content(self) -> Optional[bytes]:
  Returns:
  bytes: the file content
  """
+
+ #
+ # If the content is already loaded, use it
+ #
+ if self.__content:
+ return self.__content
+
+ #
+ # If a file has been provided, but not read yet, do it now
+ #
  if not self.file_source:
  return None
 
  data = self.file_source.open("rb").read()
  if len(data) > self.size:
  raise AttributeError("Insufficient space")
- return data
+
+ self.__content = data
+ return self.__content
 
  def __str__(self) -> str:
  return (

pyproject.toml

@@ -47,6 +47,7 @@ dependencies = [
  "pre-commit",
  "prompt_toolkit",
  "PyQt6",
+ "udmp_parser>=0.4.2",
 ]
 
 [project.optional-dependencies]