diff --git a/docs/api/process.rst b/docs/api/process.rst index ba81d44..094000d 100644 --- a/docs/api/process.rst +++ b/docs/api/process.rst @@ -17,6 +17,24 @@ Module :undoc-members: :show-inheritance: +Linux (platform-specific) +------------------------- + +.. autoclass:: memobj.process.linux.process.LinuxProcess + :members: + :undoc-members: + :show-inheritance: + +.. autoclass:: memobj.process.linux.process.LinuxMemoryRegion + :members: + :undoc-members: + :show-inheritance: + +.. autoclass:: memobj.process.linux.module.LinuxModule + :members: + :undoc-members: + :show-inheritance: + Windows (platform-specific) --------------------------- diff --git a/memobj/__init__.py b/memobj/__init__.py index d409579..6d3c8bb 100644 --- a/memobj/__init__.py +++ b/memobj/__init__.py @@ -2,7 +2,7 @@ from .allocation import Allocation, Allocator from .object import MemoryObject -from .process import Process, WindowsProcess +from .process import Process, LinuxProcess, WindowsProcess from . import property logger = logging.getLogger(__name__) @@ -17,6 +17,7 @@ "Allocator", "MemoryObject", "Process", + "LinuxProcess", "WindowsProcess", "property", ] diff --git a/memobj/hook.py b/memobj/hook.py index 34ee083..6908ec4 100644 --- a/memobj/hook.py +++ b/memobj/hook.py @@ -282,7 +282,7 @@ def hook(self): bitness = 64 if self.process.process_64_bit else 32 allocation_size = sum(map(len, hook_instructions)) - hook_allocation = self.allocate_variable("hook_site", allocation_size) + hook_allocation = self.allocate_variable("hook_site", allocation_size, preferred_start=target_address) hook_code = instructions_to_code( hook_instructions, hook_allocation.address, bitness=bitness ) @@ -351,7 +351,7 @@ def get_hook_tail(self, jump_address: int) -> tuple[list[Instruction], int]: else: bitness = 32 - decoder = Decoder(bitness, search_bytes, ip=0) + decoder = Decoder(bitness, search_bytes, ip=jump_address) for instruction in decoder: # NOTE: this is not a None check, Instruction has special bool() handling @@ -377,11 +377,10 @@ def get_hook_tail(self, jump_address: int) -> tuple[list[Instruction], int]: f"Original code contains a far branch: {instruction}" ) - # TODO: try and fix the jump instead - if not near_target < self._jump_needed - position: - raise ValueError( - f"Original code contains a near jump outside of captured code: {instruction}" - ) + # Near branches are handled by BlockEncoder: + # - Branches to instructions within this set: automatically linked + # - Branches outside this set: encoded to the original absolute target + # (safe because the original code at that address is not modified) # TODO: figure out how xbegin works case FlowControl.XBEGIN_XABORT_XEND: diff --git a/memobj/process/__init__.py b/memobj/process/__init__.py index 2129343..d9e6947 100644 --- a/memobj/process/__init__.py +++ b/memobj/process/__init__.py @@ -1,7 +1,9 @@ from .base import Process from .module import Module +from .linux.module import LinuxModule +from .linux.process import LinuxProcess from .windows.module import WindowsModule from .windows.process import WindowsProcess -__all__ = ["Process", "Module", "WindowsModule", "WindowsProcess"] +__all__ = ["Process", "Module", "LinuxModule", "LinuxProcess", "WindowsModule", "WindowsProcess"] diff --git a/memobj/process/linux/__init__.py b/memobj/process/linux/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/memobj/process/linux/module.py b/memobj/process/linux/module.py new file mode 100644 index 0000000..1fd6669 --- /dev/null +++ b/memobj/process/linux/module.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +import subprocess +from pathlib import Path +from typing import TYPE_CHECKING, Self + +from memobj.process.module import Module + +if TYPE_CHECKING: + from .process import LinuxProcess + + +class LinuxModule(Module): + _symbols: dict[str, int] | None = None + + @classmethod + def from_name( + cls, process: LinuxProcess, name: str, *, ignore_case: bool = True + ) -> Self: + if ignore_case: + name = name.lower() + + for module in cls.get_all_modules(process): + compare = module.name.lower() if ignore_case else module.name + if compare == name: + return module + + raise ValueError(f"No module named {name!r}") + + @classmethod + def get_all_modules(cls, process: LinuxProcess) -> list[Self]: + """Build module list from /proc/{pid}/maps, merging segments of the same file.""" + seen: dict[str, tuple[int, int]] = {} # path -> (min_base, max_end) + + with open(f"/proc/{process.process_id}/maps") as f: + for line in f: + parts = line.split() + if len(parts) < 6: + continue + + pathname = parts[5] + if not pathname or pathname.startswith("["): + continue + + start_str, end_str = parts[0].split("-") + start = int(start_str, 16) + end = int(end_str, 16) + + if pathname in seen: + existing_base, existing_end = seen[pathname] + seen[pathname] = (min(existing_base, start), max(existing_end, end)) + else: + seen[pathname] = (start, end) + + return [ + cls( + name=Path(path).name, + base_address=base, + executable_path=path, + size=end - base, + process=process, + ) + for path, (base, end) in seen.items() + ] + + def _is_position_independent(self) -> bool: + """Return True for PIE executables and shared libraries (ELF type ET_DYN).""" + try: + with open(self.executable_path, "rb") as f: + header = f.read(18) + if len(header) < 18 or header[:4] != b"\x7fELF": + return False + endian = "little" if header[5] == 1 else "big" + e_type = int.from_bytes(header[16:18], endian) + return e_type == 3 # ET_DYN + except OSError: + return False + + def get_symbols(self) -> dict[str, int]: + if self._symbols is not None: + return self._symbols + + is_pie = self._is_position_independent() + symbols: dict[str, int] = {} + + # Try both dynamic (-D) and static symbol tables; merge, with static + # taking precedence for address accuracy on plain executables. + for nm_args in ( + ["nm", "--defined-only", "-P", self.executable_path], + ["nm", "-D", "--defined-only", "-P", self.executable_path], + ): + try: + result = subprocess.run( + nm_args, + capture_output=True, + text=True, + check=False, + ) + except FileNotFoundError: + break + + for line in result.stdout.splitlines(): + # nm -P format: name type value size + parts = line.split() + if len(parts) < 3: + continue + sym_name = parts[0].split("@")[0] # strip GLIBC version suffixes + sym_type = parts[1] + if sym_type in ("U", "w", "v"): + continue + try: + offset = int(parts[2], 16) + # For PIE binaries and shared libs, nm gives load-relative offsets + addr = self.base_address + offset if is_pie else offset + if sym_name not in symbols: + symbols[sym_name] = addr + except ValueError: + continue + + self._symbols = symbols + return symbols + + def get_symbol_with_name(self, name: str) -> int: + try: + return self.get_symbols()[name] + except KeyError: + raise ValueError(f"No symbol named {name!r}") diff --git a/memobj/process/linux/process.py b/memobj/process/linux/process.py new file mode 100644 index 0000000..42486b2 --- /dev/null +++ b/memobj/process/linux/process.py @@ -0,0 +1,786 @@ +from __future__ import annotations + +import ctypes +import ctypes.util +import functools +import os +import struct +from dataclasses import dataclass +from pathlib import Path +from typing import Self + +import regex + +from memobj.process.base import Process +from memobj.process.linux.module import LinuxModule + +_libc_name = ctypes.util.find_library("c") +_libc = ctypes.CDLL(_libc_name, use_errno=True) if _libc_name else None + +_PROT_READ = 0x1 +_PROT_WRITE = 0x2 +_PROT_EXEC = 0x4 +_MAP_PRIVATE = 0x2 +_MAP_ANONYMOUS = 0x20 +_MAP_FAILED: int = ctypes.c_size_t(-1).value + +_PTRACE_PEEKTEXT = 1 +_PTRACE_POKETEXT = 4 +_PTRACE_CONT = 7 +_PTRACE_GETREGS = 12 +_PTRACE_SETREGS = 13 +_PTRACE_ATTACH = 16 +_PTRACE_DETACH = 17 + +_RTLD_NOW = 2 +_MMAP_SYSCALL = 9 +_MUNMAP_SYSCALL = 11 +_MAP_FIXED_NOREPLACE = 0x100000 + + +class _UserRegsStruct(ctypes.Structure): + _fields_ = [ + ("r15", ctypes.c_ulonglong), + ("r14", ctypes.c_ulonglong), + ("r13", ctypes.c_ulonglong), + ("r12", ctypes.c_ulonglong), + ("rbp", ctypes.c_ulonglong), + ("rbx", ctypes.c_ulonglong), + ("r11", ctypes.c_ulonglong), + ("r10", ctypes.c_ulonglong), + ("r9", ctypes.c_ulonglong), + ("r8", ctypes.c_ulonglong), + ("rax", ctypes.c_ulonglong), + ("rcx", ctypes.c_ulonglong), + ("rdx", ctypes.c_ulonglong), + ("rsi", ctypes.c_ulonglong), + ("rdi", ctypes.c_ulonglong), + ("orig_rax", ctypes.c_ulonglong), + ("rip", ctypes.c_ulonglong), + ("cs", ctypes.c_ulonglong), + ("eflags", ctypes.c_ulonglong), + ("rsp", ctypes.c_ulonglong), + ("ss", ctypes.c_ulonglong), + ("fs_base", ctypes.c_ulonglong), + ("gs_base", ctypes.c_ulonglong), + ("ds", ctypes.c_ulonglong), + ("es", ctypes.c_ulonglong), + ("fs", ctypes.c_ulonglong), + ("gs", ctypes.c_ulonglong), + ] + + +@dataclass +class LinuxMemoryRegion: + """Memory region info from /proc/{pid}/maps (Linux analogue of MEMORY_BASIC_INFORMATION).""" + + base_address: int + size: int + permissions: str + offset: int + device: str + inode: int + pathname: str + + @property + def readable(self) -> bool: + return "r" in self.permissions + + @property + def writable(self) -> bool: + return "w" in self.permissions + + @property + def executable(self) -> bool: + return "x" in self.permissions + + +def _ptrace(request: int, pid: int, addr: int = 0, data: int = 0) -> int: + """Low-level ptrace call. Raises OSError on failure (except PEEK which needs special handling).""" + assert _libc is not None + _libc.ptrace.restype = ctypes.c_long + _libc.ptrace.argtypes = [ + ctypes.c_ulong, ctypes.c_ulong, ctypes.c_ulong, ctypes.c_ulong + ] + result = _libc.ptrace(request, pid, addr, data) + return int(result) + + +def _ptrace_peek(pid: int, addr: int) -> int: + """Peek a word from target process memory. Returns unsigned 64-bit value.""" + assert _libc is not None + ctypes.set_errno(0) + _libc.ptrace.restype = ctypes.c_long + _libc.ptrace.argtypes = [ + ctypes.c_ulong, ctypes.c_ulong, ctypes.c_ulong, ctypes.c_ulong + ] + result = _libc.ptrace(_PTRACE_PEEKTEXT, pid, addr, 0) + err = ctypes.get_errno() + if result == -1 and err: + raise OSError(err, os.strerror(err), f"PTRACE_PEEKTEXT at {hex(addr)}") + return result & 0xFFFFFFFF_FFFFFFFF + + +def _ptrace_poke(pid: int, addr: int, word: int) -> None: + """Poke a word into target process memory.""" + ret = _ptrace(_PTRACE_POKETEXT, pid, addr, word) + if ret != 0: + err = ctypes.get_errno() + raise OSError(err, os.strerror(err), f"PTRACE_POKETEXT at {hex(addr)}") + + +def _peek_bytes(pid: int, addr: int, n_bytes: int) -> bytes: + n_words = (n_bytes + 7) // 8 + result = bytearray() + for i in range(n_words): + result.extend(struct.pack(" None: + pad = (-len(data)) % 8 + padded = data + b"\x00" * pad + for i in range(0, len(padded), 8): + word = struct.unpack(" size + + @functools.cached_property + def process_id(self) -> int: + return self._pid + + @functools.cached_property + def process_64_bit(self) -> bool: + # ELF header byte 4 (EI_CLASS): 1=ELFCLASS32, 2=ELFCLASS64 + with open(self.executable_path, "rb") as f: + header = f.read(5) + if len(header) < 5 or header[:4] != b"\x7fELF": + raise ValueError(f"Not an ELF binary: {self.executable_path}") + return header[4] == 2 + + @functools.cached_property + def executable_path(self) -> Path: + return Path(os.readlink(f"/proc/{self._pid}/exe")) + + @classmethod + def from_name(cls, name: str, *, ignore_case: bool = True) -> Self: + """ + Open a process by name, searching ``/proc//comm``. + + Args: + name: The process comm name to search for + ignore_case: Whether to do case-insensitive matching + + Returns: + The first matching LinuxProcess + """ + if ignore_case: + name = name.lower() + + for entry in Path("/proc").iterdir(): + if not entry.name.isdigit(): + continue + try: + comm = (entry / "comm").read_text().strip() + if ignore_case: + comm = comm.lower() + if comm == name: + return cls(int(entry.name)) + except (PermissionError, FileNotFoundError, ProcessLookupError): + continue + + raise ValueError(f"No process found named {name!r}") + + @classmethod + def from_id(cls, pid: int) -> Self: + """ + Open a process by PID. + + Args: + pid: The process ID + + Returns: + A LinuxProcess for the given PID + """ + if not Path(f"/proc/{pid}").exists(): + raise ValueError(f"No process with pid {pid}") + return cls(pid) + + def _ptrace_exec(self, shellcode: bytes, *, stack_data: bytes | None = None) -> int: + """ + Execute shellcode in the target process via ptrace. + + Attaches to the target, saves CPU state, writes shellcode to an + executable region (restoring it after), executes until int3, and + returns the value of RAX. If stack_data is provided it is written + below the current RSP and its address is passed as the first + argument (RDI) via the shellcode's built-in setup. + + The caller is responsible for building shellcode that ends with + ``\\xCC`` (int3). + """ + # Attach and stop the target + ret = _ptrace(_PTRACE_ATTACH, self._pid) + if ret < 0: + err = ctypes.get_errno() + raise OSError(err, os.strerror(err), "PTRACE_ATTACH failed") + os.waitpid(self._pid, 0) + + try: + regs = _UserRegsStruct() + _ptrace(_PTRACE_GETREGS, self._pid, 0, ctypes.addressof(regs)) + + # Optionally write stack_data below RSP (red zone + buffer) + if stack_data is not None: + stack_addr = (regs.rsp - 512 - len(stack_data)) & ~0xF + _poke_bytes(self._pid, stack_addr, stack_data) + + # Find an executable region large enough for the shellcode + exec_addr: int | None = None + for start, end, perms, _off, _dev, _ino, _path in self._iter_maps(): + if "x" in perms and (end - start) >= len(shellcode) + 8: + exec_addr = start + break + + if exec_addr is None: + raise RuntimeError("No executable region found in target process") + + # Save the original bytes and install shellcode + orig_code = _peek_bytes(self._pid, exec_addr, len(shellcode)) + _poke_bytes(self._pid, exec_addr, shellcode) + + # Redirect execution to our shellcode; suppress any pending syscall + # restart so the kernel honours our new RIP rather than re-entering + # the interrupted syscall (e.g. nanosleep with ERESTART_RESTARTBLOCK). + new_regs = _UserRegsStruct.from_buffer_copy(bytearray(regs)) + new_regs.rip = exec_addr + new_regs.orig_rax = 0xFFFFFFFF_FFFFFFFF + _ptrace(_PTRACE_SETREGS, self._pid, 0, ctypes.addressof(new_regs)) + + # Run until int3 (SIGTRAP / signal 5) + _ptrace(_PTRACE_CONT, self._pid, 0, 0) + os.waitpid(self._pid, 0) + + # Read result from RAX + result_regs = _UserRegsStruct() + _ptrace(_PTRACE_GETREGS, self._pid, 0, ctypes.addressof(result_regs)) + rax = result_regs.rax + + finally: + # Always restore original code and registers before detaching + try: + _poke_bytes(self._pid, exec_addr, orig_code) + _ptrace(_PTRACE_SETREGS, self._pid, 0, ctypes.addressof(regs)) + except Exception: + pass + _ptrace(_PTRACE_DETACH, self._pid, 0, 0) + + return rax + + def _find_remote_dlopen(self) -> int: + """ + Return the virtual address of ``dlopen`` in the target process. + + Finds libc (or libdl) in the target's maps, then uses ``nm -D`` to + look up dlopen's load-relative offset in that library file and adds it + to the library's runtime base address. This works even when Python + and the target binary link against different builds of the same library + (e.g. different Nix store paths for the same glibc version). + """ + import subprocess + import time + + # Find libc or libdl in the target's maps. Prefer libdl.so when present + # (older glibc keeps dlopen there); otherwise fall back to libc.so. + # Retry briefly to handle the window between process start and libc load. + target_lib_path: str | None = None + target_lib_base: int | None = None + + for _attempt in range(20): + for start, _end, _perms, file_offset, _dev, _inode, pathname in self._iter_maps(): + if not pathname or pathname.startswith("["): + continue + basename = Path(pathname).name + if file_offset != 0: + continue + if basename.startswith("libdl.so"): + target_lib_path = pathname + target_lib_base = start + break + if basename == "libc.so.6" and target_lib_path is None: + target_lib_path = pathname + target_lib_base = start + if target_lib_path is not None: + break + time.sleep(0.05) + + if target_lib_path is None or target_lib_base is None: + raise RuntimeError( + f"Could not find libc/libdl in target process {self._pid} maps" + ) + + # Use nm to get dlopen's load-relative offset from the library file. + # nm -P format: "name type value size" (value is hex, no "0x" prefix). + # Symbol names may have version suffixes like "dlopen@GLIBC_2.34". + result = subprocess.run( + ["nm", "-D", "--defined-only", "-P", target_lib_path], + capture_output=True, + text=True, + check=False, + ) + + dlopen_offset: int | None = None + for line in result.stdout.splitlines(): + parts = line.split() + if len(parts) < 3: + continue + sym_name = parts[0].split("@")[0] # strip version suffix + if sym_name == "dlopen": + try: + dlopen_offset = int(parts[2], 16) + break + except ValueError: + continue + + if dlopen_offset is None: + raise RuntimeError( + f"Could not find dlopen symbol in {target_lib_path}. " + "Is libdl/libc present with exported dynamic symbols?" + ) + + return target_lib_base + dlopen_offset + + def inject_so(self, path: Path) -> bool: + """ + Inject a shared library into this process via ptrace. + + Uses ptrace to temporarily hijack the target process, writes the + .so path onto its stack, then calls ``dlopen(path, RTLD_NOW)``. + Requires the caller to be the parent of the target (or have + CAP_SYS_PTRACE) and the process to be 64-bit x86. + + Args: + path: Absolute path to the ``.so`` file to inject + + Returns: + True if dlopen returned a non-NULL handle + """ + path_bytes = str(path.resolve()).encode() + b"\x00" + dlopen_addr = self._find_remote_dlopen() + + # We write path_bytes via stack_data; its address on the stack is + # (rsp - 512 - len(path_bytes)) & ~0xF — we need it in the shellcode. + # Instead, we build a two-stage approach: first write path_bytes via + # PTRACE_POKETEXT, then build a shellcode that knows that address. + # + # To keep it simple, we compute the path address inside _ptrace_exec: + # stack_addr = (rsp - 512 - len(path_bytes)) & ~0xF + # Since we know rsp after PTRACE_GETREGS, we need a two-pass approach. + # We use a callback-style helper instead. + + # Do everything inline (cannot use _ptrace_exec directly because we + # need rsp to compute path_addr before building the shellcode). + ret = _ptrace(_PTRACE_ATTACH, self._pid) + if ret < 0: + err = ctypes.get_errno() + raise OSError(err, os.strerror(err), "PTRACE_ATTACH failed") + os.waitpid(self._pid, 0) + + exec_addr: int | None = None + orig_code: bytes = b"" + regs = _UserRegsStruct() + + try: + _ptrace(_PTRACE_GETREGS, self._pid, 0, ctypes.addressof(regs)) + + # Write path string well below the current stack pointer. + # dlopen's implementation chains many functions and can consume several + # KB of stack — we need the path string to survive the entire call. + # 4096 bytes of clearance is enough for any glibc version in practice. + path_addr = (regs.rsp - 4096 - len(path_bytes)) & ~0xF + _poke_bytes(self._pid, path_addr, path_bytes) + + # Build shellcode: align stack, call dlopen(path_addr, RTLD_NOW), int3. + # x86-64 ABI requires RSP % 16 == 0 at the point of a CALL instruction + # (before CALL pushes the 8-byte return address). We use AND to mask the + # lower 4 bits without using a scratch register. + shellcode = ( + b"\x48\x83\xE4\xF0" + # and rsp, ~15 + b"\x48\xBF" + struct.pack("= len(shellcode) + 8: + exec_addr = start + break + + if exec_addr is None: + raise RuntimeError("No executable region found in target process") + + orig_code = _peek_bytes(self._pid, exec_addr, len(shellcode)) + _poke_bytes(self._pid, exec_addr, shellcode) + + new_regs = _UserRegsStruct.from_buffer_copy(bytearray(regs)) + new_regs.rip = exec_addr + # Prevent the kernel from restarting a pending syscall (e.g. nanosleep). + # If orig_rax != -1 the kernel would restart the interrupted syscall + # instead of jumping to our RIP when we call PTRACE_CONT. + new_regs.orig_rax = 0xFFFFFFFF_FFFFFFFF + _ptrace(_PTRACE_SETREGS, self._pid, 0, ctypes.addressof(new_regs)) + + import signal as _signal + _ptrace(_PTRACE_CONT, self._pid, 0, 0) + _, wstatus = os.waitpid(self._pid, 0) + + if not os.WIFSTOPPED(wstatus) or os.WSTOPSIG(wstatus) != _signal.SIGTRAP: + sig = os.WSTOPSIG(wstatus) if os.WIFSTOPPED(wstatus) else -1 + raise RuntimeError( + f"inject_so shellcode did not hit int3 (got signal {sig}); " + "dlopen likely crashed — check RSP alignment or .so path" + ) + + result_regs = _UserRegsStruct() + _ptrace(_PTRACE_GETREGS, self._pid, 0, ctypes.addressof(result_regs)) + success = result_regs.rax != 0 + + finally: + if exec_addr is not None and orig_code: + try: + _poke_bytes(self._pid, exec_addr, orig_code) + except Exception: + pass + try: + _ptrace(_PTRACE_SETREGS, self._pid, 0, ctypes.addressof(regs)) + except Exception: + pass + _ptrace(_PTRACE_DETACH, self._pid, 0, 0) + + return success + + def _remote_mmap(self, size: int, *, preferred_start: int | None = None) -> int: + """Allocate memory in the remote process by injecting a mmap syscall.""" + addr_arg = preferred_start if preferred_start is not None else 0 + flags = _MAP_PRIVATE | _MAP_ANONYMOUS + prot = _PROT_READ | _PROT_WRITE | _PROT_EXEC + + # syscall calling convention: rax=num, rdi, rsi, rdx, r10, r8, r9 + # (4th syscall arg uses r10, not rcx) + shellcode = ( + b"\x48\x83\xE4\xF0" + # and rsp, ~15 + b"\x48\xBF" + struct.pack(" int: + """Allocate rwx memory within ±2GB of near_addr using MAP_FIXED_NOREPLACE.""" + page_size = 0x1000 + aligned_size = (size + page_size - 1) & ~(page_size - 1) + two_gb = 0x80000000 + + lo = max(page_size, near_addr - two_gb) + hi = min(0xFFFF_FFFF_FFFF_F000, near_addr + two_gb) + + # Collect sorted list of [start, end) occupied regions + occupied: list[tuple[int, int]] = [] + with open(f"/proc/{self._pid}/maps") as f: + for line in f: + parts = line.split() + if not parts: + continue + start_s, end_s = parts[0].split("-") + occupied.append((int(start_s, 16), int(end_s, 16))) + occupied.sort() + + # Walk candidate addresses from near_addr outward, try gaps + def _gaps(): + prev_end = 0 + for start, end in occupied: + if prev_end < start: + gap_lo = max(prev_end, lo) + gap_hi = min(start, hi) + if gap_hi - gap_lo >= aligned_size: + yield gap_lo, gap_hi + prev_end = max(prev_end, end) + # After last region + tail_lo = max(prev_end, lo) + if hi - tail_lo >= aligned_size: + yield tail_lo, hi + + # Sort gaps by distance from near_addr + gaps = sorted(_gaps(), key=lambda g: min(abs(g[0] - near_addr), abs(g[1] - near_addr))) + + prot = _PROT_READ | _PROT_WRITE | _PROT_EXEC + flags = _MAP_PRIVATE | _MAP_ANONYMOUS | _MAP_FIXED_NOREPLACE + + for gap_lo, gap_hi in gaps: + # Try at the end of the gap closest to near_addr + candidates = [gap_lo, gap_hi - aligned_size] + for candidate in candidates: + candidate = candidate & ~(page_size - 1) + if candidate < gap_lo or candidate + aligned_size > gap_hi: + continue + if not (lo <= candidate and candidate + aligned_size <= hi): + continue + shellcode = ( + b"\x48\x83\xE4\xF0" + + b"\x48\xBF" + struct.pack(" None: + """Free memory in the remote process by injecting a munmap syscall.""" + shellcode = ( + b"\x48\xBF" + struct.pack(" int: + if self._pid != os.getpid(): + if preferred_start is not None: + return self._remote_mmap_near(size, preferred_start) + return self._remote_mmap(size) + + assert _libc is not None + _libc.mmap.restype = ctypes.c_size_t + _libc.mmap.argtypes = [ + ctypes.c_void_p, + ctypes.c_size_t, + ctypes.c_int, + ctypes.c_int, + ctypes.c_int, + ctypes.c_long, + ] + addr = _libc.mmap( + ctypes.c_void_p(preferred_start), + size, + _PROT_READ | _PROT_WRITE | _PROT_EXEC, + _MAP_PRIVATE | _MAP_ANONYMOUS, + -1, + 0, + ) + if addr == _MAP_FAILED: + errno_val = ctypes.get_errno() + raise OSError(errno_val, os.strerror(errno_val), f"mmap failed for size {size}") + self._allocations[addr] = size + return addr + + def free_memory(self, address: int): + size = self._allocations.pop(address, None) + if size is None: + raise ValueError(f"No tracked allocation at {hex(address)}") + + if self._pid != os.getpid(): + self._remote_munmap(address, size) + return + + assert _libc is not None + _libc.munmap.restype = ctypes.c_int + _libc.munmap.argtypes = [ctypes.c_void_p, ctypes.c_size_t] + result = _libc.munmap(ctypes.c_void_p(address), size) + if result != 0: + errno_val = ctypes.get_errno() + raise OSError(errno_val, os.strerror(errno_val), f"munmap failed for {hex(address)}") + + def read_memory(self, address: int, size: int) -> bytes: + fd = os.open(f"/proc/{self._pid}/mem", os.O_RDONLY) + try: + data = os.pread(fd, size, address) + finally: + os.close(fd) + + if len(data) != size: + raise OSError( + f"Short read at {hex(address)}: expected {size} bytes, got {len(data)}" + ) + return data + + def write_memory(self, address: int, value: bytes): + if self._pid == os.getpid(): + ctypes.memmove(address, value, len(value)) + return + + # Try /proc/pid/mem first (works for rw/rwx pages without ptrace overhead) + try: + fd = os.open(f"/proc/{self._pid}/mem", os.O_WRONLY) + try: + written = os.pwrite(fd, value, address) + finally: + os.close(fd) + if written != len(value): + raise OSError( + f"Short write at {hex(address)}: wrote {written} of {len(value)}" + ) + return + except OSError: + pass + + # Fall back to PTRACE_POKETEXT for r-xp pages (e.g. patching text section) + ret = _ptrace(_PTRACE_ATTACH, self._pid) + if ret < 0: + err = ctypes.get_errno() + raise OSError(err, os.strerror(err), "PTRACE_ATTACH failed") + os.waitpid(self._pid, 0) + try: + _poke_bytes(self._pid, address, value) + finally: + _ptrace(_PTRACE_DETACH, self._pid, 0, 0) + + def scan_memory( + self, + pattern: regex.Pattern[bytes] | bytes, + *, + module: str | LinuxModule | None = None, + ) -> list[int]: + if isinstance(pattern, bytes): + compiled = regex.compile(regex.escape(pattern), regex.DOTALL) + else: + compiled = pattern + + module_name: str | None = None + if module is not None: + module_name = module if isinstance(module, str) else module.name + + matches: list[int] = [] + for start, end, perms, _, _, _, pathname in self._iter_maps(): + if "r" not in perms: + continue + + if module_name is not None: + region_basename = Path(pathname).name if pathname else "" + if region_basename.lower() != module_name.lower(): + continue + + try: + data = self.read_memory(start, end - start) + except OSError: + continue + + for match in regex.finditer(compiled, data): + matches.append(start + match.span()[0]) + + return matches + + def _iter_maps(self): + """Parse /proc/{pid}/maps, yielding (start, end, perms, offset, device, inode, pathname).""" + with open(f"/proc/{self._pid}/maps") as f: + for line in f: + parts = line.split() + if not parts: + continue + addr_range = parts[0] + perms = parts[1] if len(parts) > 1 else "----" + offset = int(parts[2], 16) if len(parts) > 2 else 0 + device = parts[3] if len(parts) > 3 else "00:00" + inode = int(parts[4]) if len(parts) > 4 else 0 + pathname = parts[5] if len(parts) > 5 else "" + + start_str, end_str = addr_range.split("-") + yield int(start_str, 16), int(end_str, 16), perms, offset, device, inode, pathname + + # note: platform dependent + def virtual_query(self, address: int = 0) -> LinuxMemoryRegion: + """ + Get information about the memory region containing address. + Linux equivalent of VirtualQueryEx on Windows. + + Args: + address: An address within the region to query (default 0 gives the first region) + + Returns: + LinuxMemoryRegion describing the region + """ + for start, end, perms, offset, device, inode, pathname in self._iter_maps(): + if start <= address < end: + return LinuxMemoryRegion( + base_address=start, + size=end - start, + permissions=perms, + offset=offset, + device=device, + inode=inode, + pathname=pathname, + ) + raise ValueError(f"No memory region found at address {hex(address)}") + + # note: platform dependent + def get_modules(self, base_only: bool = False) -> list[LinuxModule] | LinuxModule: + """ + Get loaded modules from /proc/{pid}/maps. + + Args: + base_only: If True, return only the main executable module + + Returns: + List of LinuxModule, or a single LinuxModule if base_only=True + """ + all_modules = LinuxModule.get_all_modules(self) + + if not base_only: + return all_modules + + exe_path = str(self.executable_path) + for module in all_modules: + if module.executable_path == exe_path: + return module + + raise ValueError(f"Base module not found for {exe_path}") + + # note: platform dependent + def get_module_named(self, name: str, *, ignore_case: bool = True) -> LinuxModule: + """ + Find a loaded module by filename. + + Args: + name: The module filename (basename) to search for + ignore_case: Whether to do case-insensitive matching + + Returns: + The matching LinuxModule + """ + return LinuxModule.from_name(self, name, ignore_case=ignore_case) diff --git a/memobj/process/windows/module.py b/memobj/process/windows/module.py index 1a728d5..d57c82c 100644 --- a/memobj/process/windows/module.py +++ b/memobj/process/windows/module.py @@ -17,12 +17,13 @@ """ import ctypes +import time from typing import TYPE_CHECKING, Self from collections.abc import Iterator from memobj.process import Module -from .utils import CheckWindowsOsError, ModuleEntry32 +from .utils import ModuleEntry32 if TYPE_CHECKING: from .process import WindowsProcess @@ -30,6 +31,11 @@ INVALID_HANDLE_VALUE: int = -1 TH32CS_SNAPMODULE: int = 0x8 +# https://learn.microsoft.com/en-us/windows/win32/debug/system-error-codes--0-499- +ERROR_NO_MORE_FILES: int = 18 +# how long to keep retrying a snapshot of a freshly spawned process whose +# module list isn't populated yet +MODULE_LIST_RETRY_TIMEOUT: float = 5.0 class WindowsModule(Module): @@ -44,7 +50,14 @@ def _iter_modules(process: "WindowsProcess") -> Iterator[ModuleEntry32]: Note that the yielded modules are only valid for one iteration, i.e. references to them should not be stored """ - with CheckWindowsOsError(): + module_entry = ModuleEntry32() + module_entry.dwSize = ctypes.sizeof(ModuleEntry32) + + # a process that was just spawned can briefly report an empty module + # list (Module32First fails with ERROR_NO_MORE_FILES) before it has + # finished loading its modules, so retry for a bit instead of failing + deadline = time.monotonic() + MODULE_LIST_RETRY_TIMEOUT + while True: # https://learn.microsoft.com/en-us/windows/win32/api/tlhelp32/nf-tlhelp32-createtoolhelp32snapshot module_snapshot = ctypes.windll.kernel32.CreateToolhelp32Snapshot( TH32CS_SNAPMODULE, process.process_id @@ -53,17 +66,23 @@ def _iter_modules(process: "WindowsProcess") -> Iterator[ModuleEntry32]: if module_snapshot == INVALID_HANDLE_VALUE: raise ValueError("Creating module snapshot failed") - module_entry = ModuleEntry32() - module_entry.dwSize = ctypes.sizeof(ModuleEntry32) - # https://learn.microsoft.com/en-us/windows/win32/api/tlhelp32/nf-tlhelp32-module32first success = ctypes.windll.kernel32.Module32First( module_snapshot, ctypes.byref(module_entry) ) - if success == 0: - raise ValueError("Get first module failed") + if success != 0: + break + + last_error = ctypes.windll.kernel32.GetLastError() + ctypes.windll.kernel32.CloseHandle(module_snapshot) + + if last_error != ERROR_NO_MORE_FILES or time.monotonic() > deadline: + raise ctypes.WinError(last_error) + time.sleep(0.05) + + try: yield module_entry # https://learn.microsoft.com/en-us/windows/win32/api/tlhelp32/nf-tlhelp32-module32next @@ -74,7 +93,7 @@ def _iter_modules(process: "WindowsProcess") -> Iterator[ModuleEntry32]: != 0 ): yield module_entry - + finally: # https://learn.microsoft.com/en-us/windows/win32/api/handleapi/nf-handleapi-closehandle ctypes.windll.kernel32.CloseHandle(module_snapshot) diff --git a/tests/conftest.py b/tests/conftest.py index 1585cc1..5d1402d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -4,7 +4,7 @@ import pytest -from memobj import Process, WindowsProcess +from memobj import Process @pytest.fixture(scope="session") @@ -14,10 +14,13 @@ def process() -> Process: """ match sys.platform: case "win32": + from memobj import WindowsProcess return WindowsProcess.from_id(os.getpid()) + case "linux": + from memobj import LinuxProcess + return LinuxProcess.from_id(os.getpid()) case _: pytest.skip("Unsupported platform") - #raise RuntimeError("Unsupported platform") @pytest.fixture(scope="session") @@ -29,8 +32,15 @@ def test_binaries() -> tuple[Path, Path]: release_dir = library_root / "target" / "release" - exe_path = (release_dir / "inject_target.exe").resolve() - dll_path = (release_dir / "test_inject.dll").resolve() + match sys.platform: + case "win32": + exe_path = (release_dir / "inject_target.exe").resolve() + dll_path = (release_dir / "test_inject.dll").resolve() + case "linux": + exe_path = (release_dir / "inject_target").resolve() + dll_path = (release_dir / "libtest_inject.so").resolve() + case _: + pytest.skip("Unsupported platform") if not exe_path.exists() or not dll_path.exists(): pytest.skip("Test binaries not found") diff --git a/tests/manual/test_binaries/test_inject/Cargo.toml b/tests/manual/test_binaries/test_inject/Cargo.toml index 54bb5bd..a573919 100644 --- a/tests/manual/test_binaries/test_inject/Cargo.toml +++ b/tests/manual/test_binaries/test_inject/Cargo.toml @@ -6,5 +6,5 @@ edition = "2021" [lib] crate-type = ["cdylib"] -[dependencies] +[target.'cfg(windows)'.dependencies] windows = { version = "0.62.2", features = ["Win32_System_Console", "Win32_Foundation", "Win32_System_SystemServices"] } diff --git a/tests/manual/test_binaries/test_inject/src/lib.rs b/tests/manual/test_binaries/test_inject/src/lib.rs index 483f511..176c09c 100644 --- a/tests/manual/test_binaries/test_inject/src/lib.rs +++ b/tests/manual/test_binaries/test_inject/src/lib.rs @@ -2,25 +2,6 @@ use std::ffi::CStr; use std::fs::File; use std::io::Write; use std::os::raw::c_char; -use windows::Win32::System::SystemServices::DLL_PROCESS_ATTACH; -use windows::Win32::System::Console::AllocConsole; - - -#[no_mangle] -extern "system" fn DllMain( - _hinst_dll: *const u8, - _fdw_reason: u32, - _lpv_reserved: *const u8, -) -> bool { - match _fdw_reason { - DLL_PROCESS_ATTACH => unsafe { AllocConsole() }.unwrap_or(()), - _ => () - } - - // return True on successful attach - true -} - #[no_mangle] pub extern "C" fn create_file_at_path(path: *const c_char) -> bool { @@ -39,5 +20,23 @@ pub extern "C" fn create_file_at_path(path: *const c_char) -> bool { }; file.write_all(b"Injected").is_ok() +} +#[cfg(target_os = "windows")] +mod windows_entry { + use windows::Win32::System::Console::AllocConsole; + use windows::Win32::System::SystemServices::DLL_PROCESS_ATTACH; + + #[no_mangle] + extern "system" fn DllMain( + _hinst_dll: *const u8, + _fdw_reason: u32, + _lpv_reserved: *const u8, + ) -> bool { + match _fdw_reason { + DLL_PROCESS_ATTACH => unsafe { AllocConsole() }.unwrap_or(()), + _ => () + } + true + } } diff --git a/tests/manual/test_dll_injection.py b/tests/manual/test_dll_injection.py index 58c6360..e21e9f8 100644 --- a/tests/manual/test_dll_injection.py +++ b/tests/manual/test_dll_injection.py @@ -1,11 +1,12 @@ import subprocess +import sys -from iced_x86 import Register +import pytest import regex import memobj from memobj.utils import wait_for_value -from memobj.hook import create_capture_hook, RegisterCaptureSettings + def test_dll_injection(test_binaries): @@ -13,35 +14,103 @@ def test_dll_injection(test_binaries): proc = subprocess.Popen([str(exe_path)]) try: - process = memobj.WindowsProcess.from_id(proc.pid) - assert process.inject_dll(dll_path), "DLL injection failed" + if sys.platform == "win32": + process = memobj.WindowsProcess.from_id(proc.pid) + assert process.inject_dll(dll_path), "DLL injection failed" + elif sys.platform == "linux": + process = memobj.LinuxProcess.from_id(proc.pid) + assert process.inject_so(dll_path), "SO injection failed" + else: + pytest.skip("Unsupported platform") module = process.get_module_named(dll_path.name) - assert module is not None, "Failed to find injected DLL in remote process" + assert module is not None, "Failed to find injected library in remote process" finally: proc.terminate() + proc.wait() + + +def _get_uses_player_pattern(process, module_name: str) -> bytes: + """Read the first bytes of uses_player from the running process to use as a pattern.""" + module = process.get_module_named(module_name) + symbols = module.get_symbols() + uses_player_addr = symbols.get("uses_player") + if uses_player_addr is None: + pytest.skip("uses_player symbol not found in binary (compile with debug symbols?)") + # Read enough bytes to cover the hook jump (14 bytes) plus some margin + return process.read_memory(uses_player_addr, 32) def test_create_capture_hook(test_binaries): exe_path, _ = test_binaries - # TODO: allow passing addresses so we can do a symbol lookup - PlayerCaptureHook = create_capture_hook( - pattern=regex.escape(bytes.fromhex("48 83 EC 28 F3 0F 10 41 04 0F 2E 05 40 52 01 00 76 0D 8B 09 E8 17 FF FF FF 01 05 21 D0 01 00 48 83 C4 28 C3")), - module="inject_target.exe", - bitness=64, - register_captures=[RegisterCaptureSettings(Register.RCX, derefference=False)], - ) + match sys.platform: + case "win32": + from iced_x86 import Register + from memobj.hook import create_capture_hook, RegisterCaptureSettings - proc = subprocess.Popen([str(exe_path)]) - try: - process = memobj.WindowsProcess.from_id(proc.pid) - hook = PlayerCaptureHook(process) - hook.activate() - rcx_capture = hook.get_variable("RCX_capture") - address = wait_for_value(lambda: rcx_capture.read_typed(process.pointer_type), 0, inverse=True, timeout=60) + # TODO: allow passing addresses so we can do a symbol lookup + # note: the two 4 byte gaps are RIP-relative displacements to global data + # (the health constant and the MUTATED static) which shift whenever the + # binary is rebuilt, so they're wildcarded instead of hardcoded + pattern_prefix = regex.escape(bytes.fromhex("48 83 EC 28 F3 0F 10 41 04 0F 2E 05")) + pattern_mid = regex.escape(bytes.fromhex("76 0D 8B 09 E8 17 FF FF FF 01 05")) + pattern_suffix = regex.escape(bytes.fromhex("48 83 C4 28 C3")) + PlayerCaptureHook = create_capture_hook( + pattern=pattern_prefix + b".{4}" + pattern_mid + b".{4}" + pattern_suffix, + module="inject_target.exe", + bitness=64, + register_captures=[RegisterCaptureSettings(Register.RCX, derefference=False)], + ) - assert address != 0 - finally: - proc.terminate() + proc = subprocess.Popen([str(exe_path)]) + try: + process = memobj.WindowsProcess.from_id(proc.pid) + hook = PlayerCaptureHook(process) + hook.activate() + rcx_capture = hook.get_variable("RCX_capture") + address, _ = wait_for_value( + lambda: rcx_capture.read_typed(process.pointer_type), 0, + inverse=True, timeout=60, + ) + assert address != 0 + finally: + proc.terminate() + proc.wait() + + case "linux": + from iced_x86 import Register + from memobj.hook import create_capture_hook, RegisterCaptureSettings + + proc = subprocess.Popen([str(exe_path)]) + try: + process = memobj.LinuxProcess.from_id(proc.pid) + module_name = exe_path.name # "inject_target" + + pattern = regex.compile( + regex.escape(_get_uses_player_pattern(process, module_name)), + regex.DOTALL, + ) + + PlayerCaptureHook = create_capture_hook( + pattern=pattern, + module=module_name, + bitness=64, + # System V AMD64 ABI: first argument in RDI (not RCX) + register_captures=[RegisterCaptureSettings(Register.RDI, derefference=False)], + ) + + hook = PlayerCaptureHook(process) + hook.activate() + rdi_capture = hook.get_variable("RDI_capture") + address, _ = wait_for_value( + lambda: rdi_capture.read_typed(process.pointer_type), 0, + inverse=True, timeout=60, + ) + assert address != 0 + finally: + proc.terminate() + proc.wait() + case _: + pytest.skip("Unsupported platform") diff --git a/tests/test_linux_process.py b/tests/test_linux_process.py new file mode 100644 index 0000000..bea2344 --- /dev/null +++ b/tests/test_linux_process.py @@ -0,0 +1,212 @@ +import ctypes +import os +import struct +import sys +from pathlib import Path + +import pytest + +pytestmark = pytest.mark.skipif(sys.platform != "linux", reason="Linux only") + +from memobj.process.linux.process import LinuxMemoryRegion, LinuxProcess +from memobj.process.linux.module import LinuxModule + + +@pytest.fixture(scope="module") +def proc() -> LinuxProcess: + return LinuxProcess.from_id(os.getpid()) + + +# ── construction ────────────────────────────────────────────────────────────── + +def test_from_id(proc): + assert proc.process_id == os.getpid() + + +def test_from_id_bad_pid(): + with pytest.raises(ValueError, match="No process"): + LinuxProcess.from_id(999_999_999) + + +def test_from_name(): + comm = Path(f"/proc/{os.getpid()}/comm").read_text().strip() + found = LinuxProcess.from_name(comm) + found_comm = Path(f"/proc/{found.process_id}/comm").read_text().strip() + assert found_comm == comm + + +def test_from_name_missing(): + with pytest.raises(ValueError, match="No process found"): + LinuxProcess.from_name("__definitely_no_such_process__") + + +# ── process metadata ────────────────────────────────────────────────────────── + +def test_process_id(proc): + assert proc.process_id == os.getpid() + + +def test_process_64_bit(proc): + assert proc.process_64_bit == proc.python_64_bit + + +def test_executable_path(proc): + path = proc.executable_path + assert path.exists() + assert path.is_file() + assert "python" in path.name.lower() + + +# ── memory read / write ─────────────────────────────────────────────────────── + +def test_read_memory(proc): + data = b"memobj_read_test" + buf = ctypes.create_string_buffer(data) + result = proc.read_memory(ctypes.addressof(buf), len(data)) + assert result == data + + +def test_write_memory(proc): + buf = ctypes.create_string_buffer(16) + proc.write_memory(ctypes.addressof(buf), b"\xAB" * 16) + assert buf.raw == b"\xAB" * 16 + + +def test_read_formatted(proc): + value = 1337 + buf = ctypes.c_int32(value) + result = proc.read_formatted(ctypes.addressof(buf), "=i") + assert result == value + + +def test_write_formatted(proc): + buf = ctypes.c_int32(0) + proc.write_formatted(ctypes.addressof(buf), "=i", 9999) + assert buf.value == 9999 + + +# ── allocation ──────────────────────────────────────────────────────────────── + +def test_allocate_free_memory(proc): + addr = proc.allocate_memory(4096) + assert addr > 0 + proc.write_memory(addr, b"\xDE\xAD\xBE\xEF") + assert proc.read_memory(addr, 4) == b"\xDE\xAD\xBE\xEF" + proc.free_memory(addr) + + +def test_free_untracked_raises(proc): + with pytest.raises(ValueError, match="No tracked allocation"): + proc.free_memory(0x1234) + + +def test_allocate_preferred_start(proc): + # Request a hint; kernel may or may not honour it, but it must succeed + addr = proc.allocate_memory(4096, preferred_start=0x7000_0000_0000) + assert addr > 0 + proc.free_memory(addr) + + +# ── scan_memory ─────────────────────────────────────────────────────────────── + +def test_scan_memory_finds_pattern(proc): + needle = b"\xCA\xFE\xBA\xBE\xDE\xAD" + buf = ctypes.create_string_buffer(needle) + matches = proc.scan_memory(needle) + assert ctypes.addressof(buf) in matches + + +def test_scan_memory_regex(proc): + import regex as re + needle = b"\xFE\xED\xC0\xDE" + buf = ctypes.create_string_buffer(needle) + pattern = re.compile(b"\xfe\xed..", re.DOTALL) + matches = proc.scan_memory(pattern) + assert any(m == ctypes.addressof(buf) for m in matches) + + +def test_scan_memory_module_filter(proc): + # Scanning within the base module should still find things in its code range + base = proc.get_modules(True) + # The module's base address region is code; scan for the ELF magic bytes + results = proc.scan_memory(b"\x7fELF", module=base) + assert len(results) >= 1 + + +# ── virtual_query ───────────────────────────────────────────────────────────── + +def test_virtual_query_stack(proc): + local_var = ctypes.c_int(42) + addr = ctypes.addressof(local_var) + region = proc.virtual_query(addr) + assert isinstance(region, LinuxMemoryRegion) + assert region.base_address <= addr < region.base_address + region.size + assert region.readable + + +def test_virtual_query_heap(proc): + addr = proc.allocate_memory(4096) + try: + region = proc.virtual_query(addr) + assert region.readable + assert region.writable + finally: + proc.free_memory(addr) + + +def test_virtual_query_bad_address(proc): + with pytest.raises(ValueError, match="No memory region"): + proc.virtual_query(0x1) + + +# ── modules ─────────────────────────────────────────────────────────────────── + +def test_get_modules_returns_list(proc): + modules = proc.get_modules() + assert isinstance(modules, list) + assert len(modules) > 0 + assert all(isinstance(m, LinuxModule) for m in modules) + + +def test_get_modules_contains_python(proc): + modules = proc.get_modules() + names = [m.name.lower() for m in modules] + assert any("python" in n for n in names) + + +def test_get_modules_base_only(proc): + base = proc.get_modules(base_only=True) + assert isinstance(base, LinuxModule) + assert "python" in base.name.lower() + assert base.base_address > 0 + assert base.size > 0 + + +def test_get_module_named(proc): + base = proc.get_modules(base_only=True) + found = proc.get_module_named(base.name) + assert found.name == base.name + assert found.base_address == base.base_address + + +def test_get_module_named_case_insensitive(proc): + base = proc.get_modules(base_only=True) + found = proc.get_module_named(base.name.upper()) + assert found.name == base.name + + +def test_get_module_named_missing(proc): + with pytest.raises(ValueError, match="No module named"): + proc.get_module_named("__no_such_module__.so") + + +def test_module_has_libc(proc): + modules = proc.get_modules() + libc_modules = [m for m in modules if "libc" in m.name.lower()] + assert len(libc_modules) > 0 + + +def test_module_base_address_and_size(proc): + for module in proc.get_modules(): + assert module.base_address > 0 + assert module.size > 0 diff --git a/tests/test_process.py b/tests/test_process.py index 34083ed..edeec1b 100644 --- a/tests/test_process.py +++ b/tests/test_process.py @@ -1,10 +1,20 @@ +import sys + + def test_get_module_name(process): base_module = process.get_modules(True) - assert base_module.name == "python.exe" + if sys.platform == "win32": + assert base_module.name == "python.exe" + else: + assert "python" in base_module.name.lower() def test_get_module_named(process): - process.get_module_named("python.exe") + if sys.platform == "win32": + process.get_module_named("python.exe") + else: + base = process.get_modules(True) + process.get_module_named(base.name) # TODO: find a version independent way to test this diff --git a/uv.lock b/uv.lock index 3a9005a..dcd410f 100644 --- a/uv.lock +++ b/uv.lock @@ -693,11 +693,11 @@ wheels = [ [[package]] name = "pygments" -version = "2.19.2" +version = "2.20.0" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/b0/77/a5b8c569bf593b0140bde72ea885a803b82086995367bf2037de0159d924/pygments-2.19.2.tar.gz", hash = "sha256:636cb2477cec7f8952536970bc533bc43743542f70392ae026374600add5b887", size = 4968631, upload-time = "2025-06-21T13:39:12.283Z" } +sdist = { url = "https://files.pythonhosted.org/packages/c3/b2/bc9c9196916376152d655522fdcebac55e66de6603a76a02bca1b6414f6c/pygments-2.20.0.tar.gz", hash = "sha256:6757cd03768053ff99f3039c1a36d6c0aa0b263438fcab17520b30a303a82b5f", size = 4955991, upload-time = "2026-03-29T13:29:33.898Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/c7/21/705964c7812476f378728bdf590ca4b771ec72385c533964653c68e86bdc/pygments-2.19.2-py3-none-any.whl", hash = "sha256:86540386c03d588bb81d44bc3928634ff26449851e99741617ecb9037ee5ec0b", size = 1225217, upload-time = "2025-06-21T13:39:07.939Z" }, + { url = "https://files.pythonhosted.org/packages/f4/7e/a72dd26f3b0f4f2bf1dd8923c85f7ceb43172af56d63c7383eb62b332364/pygments-2.20.0-py3-none-any.whl", hash = "sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176", size = 1231151, upload-time = "2026-03-29T13:29:30.038Z" }, ] [[package]]