diff --git a/PyMemoryEditor/__init__.py b/PyMemoryEditor/__init__.py index ba2f795..05c6cea 100644 --- a/PyMemoryEditor/__init__.py +++ b/PyMemoryEditor/__init__.py @@ -10,6 +10,7 @@ __author__ = "Jean Loui Bernard Silva de Jesus" __version__ = "2.0.0" +import logging import sys from typing import TYPE_CHECKING @@ -22,6 +23,16 @@ ProcessNotFoundError, PyMemoryEditorError, ) +from .process.thread_info import ThreadInfo + + +# Package-wide logger. Silent by default (NullHandler) — embedding apps opt in +# with `logging.basicConfig(level=logging.DEBUG)` or by attaching a handler to +# the "PyMemoryEditor" logger. Backends emit DEBUG for transient skips (pages +# vanished mid-scan) and WARNING for surprising-but-recovered conditions +# (partial reads, mach_vm_protect restore failure). +logger = logging.getLogger("PyMemoryEditor") +logger.addHandler(logging.NullHandler()) if sys.platform == "win32": @@ -75,6 +86,8 @@ "ProcessNotFoundError", "PyMemoryEditorError", "ScanTypesEnum", + "ThreadInfo", "__author__", "__version__", + "logger", ) + _PLATFORM_EXPORTS diff --git a/PyMemoryEditor/app/cheat_table.py b/PyMemoryEditor/app/cheat_table.py index 5b2a4cb..6b821ce 100644 --- a/PyMemoryEditor/app/cheat_table.py +++ b/PyMemoryEditor/app/cheat_table.py @@ -17,6 +17,7 @@ """ import copy import json +import logging from typing import Dict, List, Optional, Tuple from PySide6.QtCore import Qt, QTimer @@ -55,6 +56,9 @@ # poll-interval constant from this module before the split. _TICK_INTERVAL_MS = TICK_INTERVAL_MS +# Child of "PyMemoryEditor" — the Log Console captures these via propagation. +_LOG = logging.getLogger(__name__) + class CheatTable(QWidget): """Bottom pane: saved addresses, freezing, manual edits.""" @@ -286,6 +290,14 @@ def _on_cell_changed(self, row: int, column: int) -> None: QMessageBox.critical( self, "Write Failed", f"{type(exc).__name__}: {exc}" ) + _LOG.warning( + "Cheat-table write failed at 0x%X (%s, %dB): %s: %s", + entry.address, + entry.spec.pytype.__name__, + entry.length, + type(exc).__name__, + exc, + ) return entry.last_value = value diff --git a/PyMemoryEditor/app/log_console_dialog.py b/PyMemoryEditor/app/log_console_dialog.py new file mode 100644 index 0000000..380adc2 --- /dev/null +++ b/PyMemoryEditor/app/log_console_dialog.py @@ -0,0 +1,198 @@ +# -*- coding: utf-8 -*- +""" +Log console — a live view of the ``PyMemoryEditor`` logger. + +The library logs DEBUG events when scanning helpers skip transient pages +(unmapped, no-access, partial reads) and WARNING when something +recovers-but-leaks state (macOS mach_vm_protect couldn't restore). This +dialog attaches a custom :class:`logging.Handler` to the ``PyMemoryEditor`` +logger and streams records into a read-only text view — useful to +understand what the library is doing during a noisy scan without dropping +to the terminal. + +Cross-thread emit safety: the underlying scan workers run on QThreads. The +``Handler.emit`` method runs on whichever thread called the logger, so we +hop to the UI thread via a ``Qt.QueuedConnection`` signal before touching +the widget. +""" +import logging +from typing import Optional + +from PySide6.QtCore import QObject, Qt, Signal +from PySide6.QtGui import QFont +from PySide6.QtWidgets import ( + QCheckBox, + QComboBox, + QDialog, + QHBoxLayout, + QLabel, + QPlainTextEdit, + QPushButton, + QVBoxLayout, +) + + +_LOGGER_NAME = "PyMemoryEditor" + +# Levels offered in the UI, in increasing order of severity. +_LEVELS = ( + ("DEBUG", logging.DEBUG), + ("INFO", logging.INFO), + ("WARNING", logging.WARNING), + ("ERROR", logging.ERROR), +) + + +class _QtLogSignal(QObject): + """Tiny QObject that owns the cross-thread signal. + + A ``logging.Handler`` is not itself a QObject and can't carry signals; + composing one here keeps the handler thread-safe (emit happens via Qt's + queued event loop) without inheriting from two unrelated base classes. + """ + + line_emitted = Signal(str) + + +class _QtLogHandler(logging.Handler): + """``logging.Handler`` that forwards formatted records to a Qt signal.""" + + def __init__(self) -> None: + super().__init__() + self.bridge = _QtLogSignal() + + def emit(self, record: logging.LogRecord) -> None: + try: + line = self.format(record) + except Exception: # noqa: BLE001 — never let logging crash callers + return + # The signal is connected with Qt.QueuedConnection by the dialog so + # the actual widget update happens on the UI thread. + self.bridge.line_emitted.emit(line) + + +class LogConsoleDialog(QDialog): + """Live view of the PyMemoryEditor logger.""" + + # The dialog can be opened/closed repeatedly; we attach the handler on + # open and detach on close so the library doesn't keep growing handler + # lists across sessions. + + def __init__(self, parent=None): + super().__init__(parent) + self._handler: Optional[_QtLogHandler] = None + self._logger = logging.getLogger(_LOGGER_NAME) + # Snapshot the logger's level so we can put it back on close — the + # user opening this dialog should not silently raise the global + # verbosity for code that runs after the dialog is dismissed. + self._previous_level = self._logger.level + + self.setWindowTitle("PyMemoryEditor — Log Console") + self.resize(820, 480) + + self._build_ui() + self._attach_handler(level=logging.DEBUG) + self._apply_level("DEBUG") + + def _build_ui(self) -> None: + layout = QVBoxLayout(self) + layout.setContentsMargins(14, 14, 14, 14) + layout.setSpacing(10) + + header = QLabel( + "Log Console" + "  logger = \"PyMemoryEditor\"" + ) + header.setTextFormat(Qt.RichText) + layout.addWidget(header) + + hint = QLabel( + "Records emitted by the library while this dialog is open. " + "DEBUG-level entries surface transient skips during scans; " + "WARNING entries flag recovered-but-noisy conditions." + ) + hint.setObjectName("hint") + hint.setWordWrap(True) + layout.addWidget(hint) + + bar = QHBoxLayout() + bar.setSpacing(8) + + bar.addWidget(QLabel("Level:")) + self._level_combo = QComboBox() + for label, _ in _LEVELS: + self._level_combo.addItem(label) + self._level_combo.setCurrentText("DEBUG") + self._level_combo.currentTextChanged.connect(self._apply_level) + bar.addWidget(self._level_combo) + + self._autoscroll_check = QCheckBox("Auto-scroll") + self._autoscroll_check.setChecked(True) + bar.addWidget(self._autoscroll_check) + + bar.addStretch(1) + + clear_btn = QPushButton("Clear") + clear_btn.clicked.connect(self._on_clear) + bar.addWidget(clear_btn) + + close_btn = QPushButton("Close") + close_btn.clicked.connect(self.accept) + bar.addWidget(close_btn) + + layout.addLayout(bar) + + self._console = QPlainTextEdit() + self._console.setReadOnly(True) + self._console.setFont(QFont("Menlo, Consolas, Courier New", 10)) + self._console.setLineWrapMode(QPlainTextEdit.NoWrap) + # Cap the buffer so a long-running auto-scan doesn't grow the dialog + # memory unboundedly. ~5000 lines is plenty for live debugging. + self._console.setMaximumBlockCount(5000) + layout.addWidget(self._console, 1) + + def _attach_handler(self, level: int) -> None: + handler = _QtLogHandler() + handler.setLevel(level) + handler.setFormatter( + logging.Formatter("%(asctime)s %(levelname)-7s %(message)s", "%H:%M:%S") + ) + # QueuedConnection ensures the widget update happens on the UI thread + # even when the log record was emitted from a scan worker QThread. + handler.bridge.line_emitted.connect( + self._console_append, Qt.ConnectionType.QueuedConnection + ) + self._logger.addHandler(handler) + if self._logger.level == logging.NOTSET or self._logger.level > level: + self._logger.setLevel(level) + self._handler = handler + + def _apply_level(self, label: str) -> None: + level = dict(_LEVELS).get(label, logging.DEBUG) + if self._handler is not None: + self._handler.setLevel(level) + if self._logger.level == logging.NOTSET or self._logger.level > level: + self._logger.setLevel(level) + + def _console_append(self, line: str) -> None: + self._console.appendPlainText(line) + if self._autoscroll_check.isChecked(): + sb = self._console.verticalScrollBar() + sb.setValue(sb.maximum()) + + def _on_clear(self) -> None: + self._console.clear() + + def closeEvent(self, event): # noqa: N802 — Qt naming + # Detach the handler so the library doesn't keep emitting into a + # dialog the user has dismissed. Don't lower the logger level past + # what the caller had configured before us. + if self._handler is not None: + try: + self._handler.bridge.line_emitted.disconnect() + except (RuntimeError, TypeError): + pass + self._logger.removeHandler(self._handler) + self._handler = None + self._logger.setLevel(self._previous_level) + super().closeEvent(event) diff --git a/PyMemoryEditor/app/main_window.py b/PyMemoryEditor/app/main_window.py index b9f7cf0..154f9c9 100644 --- a/PyMemoryEditor/app/main_window.py +++ b/PyMemoryEditor/app/main_window.py @@ -16,6 +16,7 @@ +------------------------------------------------------------+ """ import json +import logging import sys from typing import List, Optional, Union @@ -46,14 +47,21 @@ from ._icon import app_icon from .application import DEFAULT_THEME_ID, THEMES, apply_theme +from .cheat_entry import CheatEntry from .cheat_table import CheatTable +from .log_console_dialog import LogConsoleDialog from .memory_map_dialog import MemoryMapDialog from .memory_viewer_dialog import MemoryViewerDialog +from .pointer_chain_dialog import PointerChainDialog from .results_view import ResultsModel, ResultsView from .scan_worker import FirstScanWorker, RefineScanWorker, ScanRequest from .scanner_panel import ScannerPanel +from .threads_dialog import ThreadsDialog +# Child of "PyMemoryEditor" — the Log Console captures these via propagation. +_LOG = logging.getLogger(__name__) + # Cadence at which we poll psutil to check the target process is still alive. # 2 s is brisk enough that a dead target's cleanup happens before the user # tries to refine a scan, but slow enough to keep the cost negligible. @@ -80,6 +88,13 @@ def __init__(self, process: AbstractProcess): self.setWindowIcon(app_icon()) self.resize(1280, 780) + # Lazy slots for the new dialogs — instantiated on first open, + # cached so subsequent opens reuse the same window (matches the + # behavior of the existing memory_map dialog). + self._threads_dialog: Optional[ThreadsDialog] = None + self._pointer_chain_dialog: Optional[PointerChainDialog] = None + self._log_console_dialog: Optional[LogConsoleDialog] = None + self._build_ui() # Heartbeat — make sure the target process is still alive. If it @@ -221,6 +236,23 @@ def _build_menu_and_toolbar(self) -> None: hex_viewer_action.triggered.connect(lambda: self._open_hex_viewer(0)) tools_menu.addAction(hex_viewer_action) + threads_action = QAction("Threads…", self) + threads_action.setShortcut(QKeySequence("Ctrl+T")) + threads_action.triggered.connect(self._open_threads_dialog) + tools_menu.addAction(threads_action) + + pointer_chain_action = QAction("Resolve Pointer Chain…", self) + pointer_chain_action.setShortcut(QKeySequence("Ctrl+Shift+P")) + pointer_chain_action.triggered.connect(self._open_pointer_chain_dialog) + tools_menu.addAction(pointer_chain_action) + + tools_menu.addSeparator() + + log_console_action = QAction("Log Console…", self) + log_console_action.setShortcut(QKeySequence("Ctrl+L")) + log_console_action.triggered.connect(self._open_log_console) + tools_menu.addAction(log_console_action) + refresh_snapshot = QAction("Refresh Region Snapshot", self) refresh_snapshot.triggered.connect(self._refresh_region_snapshot) tools_menu.addAction(refresh_snapshot) @@ -234,6 +266,7 @@ def _build_menu_and_toolbar(self) -> None: toolbar.setMovable(False) toolbar.addAction(memory_map_action) toolbar.addAction(hex_viewer_action) + toolbar.addAction(pointer_chain_action) toolbar.addSeparator() toolbar.addAction(export_results) @@ -401,6 +434,14 @@ def _fill_initial_values(self, request: ScanRequest) -> None: # Don't recurse into another scan if the user has already triggered one. if self._worker is not None: return + # AOB pattern matches don't have a "current value" the way a numeric + # scan does — read_process_memory with the spec's (bytes, length=0) + # would error, and even with a non-zero length the bytes are the same + # ones the pattern already located. Skip the auto-refresh and leave + # the value column empty — the user can promote rows to the cheat + # table for a live preview there. + if request.spec.is_pattern: + return self._on_update_values(request) def _on_new_scan(self) -> None: @@ -439,6 +480,7 @@ def _on_refresh_done(self, _kept: int) -> None: self._scanner.set_has_results(self._results_model.count() > 0) def _on_worker_error(self, message: str) -> None: + _LOG.error("Scan worker error: %s", message) QMessageBox.critical(self, "Scan error", message) self._status.showMessage(message) @@ -476,6 +518,63 @@ def _on_memory_map_closed(self, _result: int) -> None: self._region_snapshot = snap self._memory_map = None + def _open_threads_dialog(self) -> None: + if self._threads_dialog is None: + self._threads_dialog = ThreadsDialog(self._process, self) + self._threads_dialog.finished.connect(self._on_threads_dialog_closed) + else: + self._threads_dialog.refresh() + self._threads_dialog.show() + self._threads_dialog.raise_() + self._threads_dialog.activateWindow() + + def _on_threads_dialog_closed(self, _result: int) -> None: + self._threads_dialog = None + + def _open_pointer_chain_dialog(self) -> None: + if self._pointer_chain_dialog is None: + self._pointer_chain_dialog = PointerChainDialog(self._process, self) + self._pointer_chain_dialog.add_to_cheat_table.connect( + self._on_pointer_chain_promote + ) + self._pointer_chain_dialog.finished.connect( + self._on_pointer_chain_dialog_closed + ) + self._pointer_chain_dialog.show() + self._pointer_chain_dialog.raise_() + self._pointer_chain_dialog.activateWindow() + + def _on_pointer_chain_dialog_closed(self, _result: int) -> None: + self._pointer_chain_dialog = None + + def _on_pointer_chain_promote( + self, address: int, spec_label: str, length: int + ) -> None: + """Promote a resolved pointer-chain address into the cheat table.""" + entry = CheatEntry( + description="", + address=int(address), + spec_label=spec_label, + length=int(length), + ) + self._cheat.add_entry(entry) + self._status.showMessage( + f"Added 0x{address:X} to cheat table (from pointer chain)." + ) + + def _open_log_console(self) -> None: + if self._log_console_dialog is None: + self._log_console_dialog = LogConsoleDialog(self) + self._log_console_dialog.finished.connect( + self._on_log_console_closed + ) + self._log_console_dialog.show() + self._log_console_dialog.raise_() + self._log_console_dialog.activateWindow() + + def _on_log_console_closed(self, _result: int) -> None: + self._log_console_dialog = None + def _open_hex_viewer(self, address: int) -> None: self._open_hex_viewer_with_size(address, 256) @@ -603,6 +702,17 @@ def _change_process(self) -> None: self._region_snapshot = None self._results_model.clear() self._scanner.set_has_results(False) + + # Tear down auxiliary dialogs that hold a reference to the old + # process — reopening them rebuilds against the new target. + for dialog_attr in ( + "_threads_dialog", + "_pointer_chain_dialog", + ): + existing = getattr(self, dialog_attr, None) + if existing is not None: + existing.close() + setattr(self, dialog_attr, None) # Replace the cheat table — old entries point at the previous process. # QSplitter has no QLayout, so we use its native replaceWidget(index). old_cheat = self._cheat diff --git a/PyMemoryEditor/app/memory_viewer_dialog.py b/PyMemoryEditor/app/memory_viewer_dialog.py index 9d88aca..466d3ce 100644 --- a/PyMemoryEditor/app/memory_viewer_dialog.py +++ b/PyMemoryEditor/app/memory_viewer_dialog.py @@ -5,6 +5,7 @@ Polls the chosen address range at a configurable interval (Cheat Engine-style "auto-refresh") so the user can watch values change live. """ +import logging from typing import Optional from PySide6.QtCore import QTimer @@ -26,6 +27,10 @@ from ._widgets import parse_hex_address +# Child of the "PyMemoryEditor" logger, so the Log Console (which attaches a +# handler to "PyMemoryEditor") picks these up via propagation. +_LOG = logging.getLogger(__name__) + _BYTES_PER_LINE = 16 @@ -153,6 +158,13 @@ def refresh(self) -> None: except Exception as exc: # noqa: BLE001 — surface every backend error self._dump.setPlainText("") self._status.setText(f"Read failed: {type(exc).__name__}: {exc}") + _LOG.warning( + "Hex viewer read failed at 0x%X (%d bytes): %s: %s", + addr, + size, + type(exc).__name__, + exc, + ) return if not isinstance(data, (bytes, bytearray)): @@ -202,6 +214,13 @@ def _write_bytes(self) -> None: QMessageBox.critical( self, "Memory Viewer", f"Write failed:\n\n{type(exc).__name__}: {exc}" ) + _LOG.warning( + "Hex viewer write failed at 0x%X (%d bytes): %s: %s", + addr, + len(data), + type(exc).__name__, + exc, + ) return self._status.setText(f"Wrote {len(data)} bytes to 0x{addr:X}.") self.refresh() diff --git a/PyMemoryEditor/app/pointer_chain_dialog.py b/PyMemoryEditor/app/pointer_chain_dialog.py new file mode 100644 index 0000000..0a623da --- /dev/null +++ b/PyMemoryEditor/app/pointer_chain_dialog.py @@ -0,0 +1,445 @@ +# -*- coding: utf-8 -*- +""" +Pointer-chain dialog — exposes ``process.resolve_pointer_chain()``. + +The intent is to be a *direct paste path* from a Cheat-Engine cheat table. +Cheat-Engine writes chains like:: + + "game.exe" + 0x10F4F4 -> [+0x0] -> [+0x158] ; HP + +This dialog asks for: + +* a **base address** in hex, +* the **list of offsets** (the bracketed steps, comma-separated, hex), +* the **pointer size** (4 for 32-bit targets, 8 for 64-bit). + +Then it walks the chain, surfaces the final address, reads the value with +the chosen value type, and offers an *"Add to cheat table"* shortcut that +emits a signal the main window picks up — so the resolved address survives +into the regular freeze/refresh loop. + +Same construction shape as the existing Memory Map / Hex Viewer dialogs +(small, self-contained, no background worker because the chain walk is +already fast). +""" +import logging +from typing import List, Optional, Tuple + +from PySide6.QtCore import Qt, Signal +from PySide6.QtGui import QFont, QGuiApplication +from PySide6.QtWidgets import ( + QCheckBox, + QComboBox, + QDialog, + QFormLayout, + QHBoxLayout, + QLabel, + QLineEdit, + QMessageBox, + QPushButton, + QScrollArea, + QSpinBox, + QToolButton, + QVBoxLayout, + QWidget, +) + +from PyMemoryEditor import AbstractProcess + +from ._widgets import parse_hex_address +from .value_types import VALUE_TYPES, ValueTypeSpec, find_spec + + +# Child of "PyMemoryEditor" — surfaced by the Log Console via propagation. +_LOG = logging.getLogger(__name__) + + +class _OffsetField(QWidget): + """One slot in the offsets chain — visually ``[+ ]``. + + Cheat-Engine notation uses ``[+0x10] -> [+0x20]`` to show a pointer + walk; we mirror that with a label-input-label triple per offset so the + "list of offsets" reads as a chain instead of as a free-form text field. + The trailing ``×`` removes this field; the parent dialog hides the + button when only one slot remains. + """ + + removed = Signal(object) # self + + def __init__(self, parent=None) -> None: + super().__init__(parent) + layout = QHBoxLayout(self) + layout.setContentsMargins(0, 0, 0, 0) + layout.setSpacing(2) + + prefix = QLabel("[+") + prefix.setObjectName("hint") + layout.addWidget(prefix) + + self.edit = QLineEdit() + self.edit.setPlaceholderText("0") + self.edit.setFixedWidth(80) + self.edit.setFont(QFont("Menlo, Consolas, Courier New", 10)) + layout.addWidget(self.edit) + + suffix = QLabel("]") + suffix.setObjectName("hint") + layout.addWidget(suffix) + + self.remove_btn = QToolButton(self) + self.remove_btn.setText("×") + self.remove_btn.setToolTip("Remove this offset") + self.remove_btn.setFixedSize(20, 20) + self.remove_btn.clicked.connect(lambda: self.removed.emit(self)) + layout.addWidget(self.remove_btn) + + def text(self) -> str: + return self.edit.text().strip() + + +class PointerChainDialog(QDialog): + """Walk a multi-level pointer chain and surface the final address.""" + + # qulonglong: addresses regularly exceed Qt's signed-32-bit default. + add_to_cheat_table = Signal( + "qulonglong", str, int + ) # (resolved_address, spec_label, length) + + def __init__(self, process: AbstractProcess, parent=None): + super().__init__(parent) + self._process = process + self._resolved_address: Optional[int] = None + + self.setWindowTitle(f"Resolve Pointer Chain — PID {process.pid}") + self.resize(640, 460) + + self._build_ui() + + def _build_ui(self) -> None: + layout = QVBoxLayout(self) + layout.setContentsMargins(14, 14, 14, 14) + layout.setSpacing(10) + + header = QLabel( + "Resolve Pointer Chain" + ) + header.setTextFormat(Qt.RichText) + layout.addWidget(header) + + hint = QLabel( + "Paste a Cheat Engine-style chain. The base address can be a static " + "offset inside the executable (module base + offset) or a known " + "pointer in memory." + ) + hint.setObjectName("hint") + hint.setWordWrap(True) + layout.addWidget(hint) + + form = QFormLayout() + form.setHorizontalSpacing(10) + form.setVerticalSpacing(8) + + self._base_edit = QLineEdit() + self._base_edit.setPlaceholderText("e.g. 0x14010F4F4 (hex)") + form.addRow("Base address:", self._base_edit) + + # Offsets row — Cheat-Engine style chain of "[+ hex ]" slots with a + # trailing "+" button to add another hop. Wrapped in a horizontal + # scroll area so deep chains (10+ levels) don't blow up the dialog + # width. + self._offset_fields: List[_OffsetField] = [] + + offsets_container = QWidget() + self._offsets_row = QHBoxLayout(offsets_container) + self._offsets_row.setContentsMargins(0, 0, 0, 0) + self._offsets_row.setSpacing(4) + + # The "+" button lives in the layout and stays at the right end; we + # insert new fields just *before* it via insertWidget(index-1, …). + self._add_offset_btn = QToolButton() + self._add_offset_btn.setText("+") + self._add_offset_btn.setToolTip("Add another offset (one more hop in the chain)") + self._add_offset_btn.setFixedSize(28, 24) + self._add_offset_btn.clicked.connect(lambda: self._add_offset_field()) + self._offsets_row.addWidget(self._add_offset_btn) + self._offsets_row.addStretch(1) + + offsets_scroll = QScrollArea() + offsets_scroll.setWidget(offsets_container) + offsets_scroll.setWidgetResizable(True) + offsets_scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarAsNeeded) + offsets_scroll.setVerticalScrollBarPolicy(Qt.ScrollBarAlwaysOff) + offsets_scroll.setFrameShape(QScrollArea.NoFrame) + offsets_scroll.setFixedHeight(40) + form.addRow("Offsets:", offsets_scroll) + + # Start with a single empty slot so the dialog isn't blank. + self._add_offset_field() + + self._ptr_size_combo = QComboBox() + self._ptr_size_combo.addItem("8 bytes (64-bit)", 8) + self._ptr_size_combo.addItem("4 bytes (32-bit)", 4) + form.addRow("Pointer size:", self._ptr_size_combo) + + # The CE-style chain assumes ``base`` is a *static slot* in the + # executable that holds a pointer (so we dereference once before + # walking offsets). Users who paste a *direct* address (e.g. from + # the Memory Map or a fresh scan) want ``base`` to be the final + # address itself — offsets in that case are struct-field offsets, + # added without any extra dereference. + self._deref_check = QCheckBox( + "Base is a pointer (dereference it, then walk offsets)" + ) + self._deref_check.setChecked(True) + self._deref_check.setToolTip( + "Checked (Cheat-Engine style): base address holds a pointer; " + "the resolver reads that pointer, then dereferences again on each " + "offset.\n\n" + "Unchecked: base is the final address itself. Offsets are added " + "without dereferencing — useful when you pasted an address from " + "the Memory Map or want a struct field at base+offset." + ) + form.addRow("", self._deref_check) + + self._value_type_combo = QComboBox() + for spec in VALUE_TYPES: + self._value_type_combo.addItem(spec.label) + form.addRow("Read value as:", self._value_type_combo) + + self._length_spin = QSpinBox() + self._length_spin.setRange(1, 1024) + self._length_spin.setValue(4) + self._length_spin.setSuffix(" bytes") + self._value_type_combo.currentTextChanged.connect(self._on_value_type_changed) + form.addRow("Length:", self._length_spin) + + layout.addLayout(form) + + # The primary/secondary/danger QSS rules add `padding: 7px 14px; + # min-height: 20px`, making those buttons taller than a plain + # QPushButton (5px 12px). Apply the same padding to the neutral + # buttons so the whole row lines up at the Resolve button's height. + _equal_height = "padding: 7px 14px; min-height: 20px;" + + button_row = QHBoxLayout() + self._resolve_btn = QPushButton("Resolve") + self._resolve_btn.setObjectName("secondary") + self._resolve_btn.setDefault(True) + self._resolve_btn.clicked.connect(self._on_resolve) + button_row.addWidget(self._resolve_btn) + + self._copy_addr_btn = QPushButton("Copy address") + self._copy_addr_btn.setStyleSheet(_equal_height) + self._copy_addr_btn.clicked.connect(self._on_copy_address) + self._copy_addr_btn.setEnabled(False) + button_row.addWidget(self._copy_addr_btn) + + self._add_to_cheat_btn = QPushButton("Add to cheat table") + self._add_to_cheat_btn.setStyleSheet(_equal_height) + self._add_to_cheat_btn.clicked.connect(self._on_add_to_cheat) + self._add_to_cheat_btn.setEnabled(False) + button_row.addWidget(self._add_to_cheat_btn) + + button_row.addStretch(1) + + close_btn = QPushButton("Close") + close_btn.setStyleSheet(_equal_height) + close_btn.clicked.connect(self.accept) + button_row.addWidget(close_btn) + layout.addLayout(button_row) + + layout.addSpacing(8) + + self._output_label = QLabel("Resolved address: —") + self._output_label.setObjectName("hint") + self._output_label.setFont(QFont("Menlo, Consolas, Courier New", 11)) + self._output_label.setTextInteractionFlags( + Qt.TextInteractionFlag.TextSelectableByMouse + ) + layout.addWidget(self._output_label) + + self._value_label = QLabel("Value: —") + self._value_label.setFont(QFont("Menlo, Consolas, Courier New", 11)) + self._value_label.setTextInteractionFlags( + Qt.TextInteractionFlag.TextSelectableByMouse + ) + layout.addWidget(self._value_label) + + layout.addStretch(1) + + # Sync default spec/length so the spin shows the right value at open. + self._on_value_type_changed(self._value_type_combo.currentText()) + + def _add_offset_field(self) -> _OffsetField: + """Append a fresh offset slot before the ``+`` button.""" + field = _OffsetField(self) + field.removed.connect(self._remove_offset_field) + # Layout order is [field, field, …, "+" btn, stretch]. New entries go + # in position (count - 2) so they land *before* the "+" button. + insert_index = max(0, self._offsets_row.count() - 2) + self._offsets_row.insertWidget(insert_index, field) + self._offset_fields.append(field) + field.edit.setFocus() + self._update_remove_buttons() + return field + + def _remove_offset_field(self, field: _OffsetField) -> None: + """Drop ``field`` from the chain — unless it's the only one left.""" + if field not in self._offset_fields: + return + if len(self._offset_fields) == 1: + # Always keep at least one slot so the user has somewhere to type; + # clearing the input is the closest "remove" we can do here. + field.edit.clear() + return + self._offset_fields.remove(field) + self._offsets_row.removeWidget(field) + field.setParent(None) + field.deleteLater() + self._update_remove_buttons() + + def _update_remove_buttons(self) -> None: + """Hide the ``×`` on the only-remaining field so the chain never collapses.""" + only_one = len(self._offset_fields) <= 1 + for field in self._offset_fields: + field.remove_btn.setVisible(not only_one) + + def _read_offsets(self) -> Optional[List[int]]: + """Collect non-empty offsets in order; return None if any one is invalid.""" + offsets: List[int] = [] + for field in self._offset_fields: + text = field.text() + if not text: + continue + parsed = parse_hex_address(text) + if parsed is None: + # parse_hex_address only handles full hex addresses with or + # without ``0x``; try a plain hex int as a fallback for tokens + # like ``"10"`` that look ambiguous (decimal vs hex). The + # whole dialog treats offsets as hex, matching Cheat Engine. + try: + parsed = int(text, 16) + except ValueError: + return None + offsets.append(parsed) + return offsets + + def _on_value_type_changed(self, label: str) -> None: + spec = find_spec(label) + if spec is None: + return + self._length_spin.setEnabled(spec.accepts_length_override) + if not spec.accepts_length_override: + self._length_spin.setValue(spec.length) + elif spec.pytype is bytes: + self._length_spin.setValue(max(4, self._length_spin.value())) + else: + self._length_spin.setValue(16) + + def _current_spec(self) -> Tuple[ValueTypeSpec, int]: + spec = find_spec(self._value_type_combo.currentText()) or VALUE_TYPES[0] + length = self._length_spin.value() if spec.accepts_length_override else spec.length + return spec, int(length) + + def _on_resolve(self) -> None: + base_text = self._base_edit.text().strip() + if not base_text: + QMessageBox.warning(self, "Resolve", "Enter a base address first.") + return + + base = parse_hex_address(base_text) + if base is None: + QMessageBox.warning( + self, + "Resolve", + "Base address must be hex (e.g. 0x14010F4F4 or 14010F4F4).", + ) + return + + offsets = self._read_offsets() + if offsets is None: + QMessageBox.warning( + self, + "Resolve", + "Every offset must be hex (e.g. 0x10, 158, 0xC8). " + "Leave a slot empty to skip it, or click × to remove it.", + ) + return + + ptr_size = int(self._ptr_size_combo.currentData()) + dereference = self._deref_check.isChecked() + + if dereference: + try: + resolved = self._process.resolve_pointer_chain( + base, offsets, ptr_size=ptr_size + ) + except Exception as exc: # noqa: BLE001 + QMessageBox.critical( + self, + "Resolve", + "Could not walk the chain — typically one of the " + "intermediate pointers is invalid. If the base is already " + "the final address (e.g. from the Memory Map), uncheck " + "\"Base is a pointer\".\n\n" + f"{type(exc).__name__}: {exc}", + ) + _LOG.warning( + "Pointer chain resolve failed (base=0x%X, offsets=%s, " + "ptr_size=%d): %s: %s", + base, + offsets, + ptr_size, + type(exc).__name__, + exc, + ) + self._resolved_address = None + self._output_label.setText("Resolved address: —") + self._value_label.setText("Value: —") + self._copy_addr_btn.setEnabled(False) + self._add_to_cheat_btn.setEnabled(False) + return + hop_summary = f"walked {len(offsets)} hop(s)" + else: + # Raw mode: just add the offsets together. No syscalls — the + # resolver semantics here are "base + sum(offsets)". + resolved = base + sum(offsets) + hop_summary = ( + "direct (no dereference)" if not offsets + else f"base + sum of {len(offsets)} offset(s)" + ) + + self._resolved_address = resolved + self._output_label.setText( + f"Resolved address: 0x{resolved:X} ({hop_summary})" + ) + + spec, length = self._current_spec() + try: + value = self._process.read_process_memory( + resolved, spec.pytype, length + ) + except Exception as exc: # noqa: BLE001 + self._value_label.setText( + f"Value: " + ) + else: + try: + formatted = spec.format(value) + except Exception: # noqa: BLE001 + formatted = repr(value) + self._value_label.setText(f"Value ({spec.label}): {formatted}") + + self._copy_addr_btn.setEnabled(True) + self._add_to_cheat_btn.setEnabled(True) + + def _on_copy_address(self) -> None: + if self._resolved_address is None: + return + QGuiApplication.clipboard().setText(f"{self._resolved_address:X}") + + def _on_add_to_cheat(self) -> None: + if self._resolved_address is None: + return + spec, length = self._current_spec() + self.add_to_cheat_table.emit(self._resolved_address, spec.label, length) diff --git a/PyMemoryEditor/app/scan_worker.py b/PyMemoryEditor/app/scan_worker.py index 029657d..73be44b 100644 --- a/PyMemoryEditor/app/scan_worker.py +++ b/PyMemoryEditor/app/scan_worker.py @@ -84,7 +84,18 @@ def __init__(self, process: AbstractProcess, request: ScanRequest, parent=None): def run(self) -> None: req = self._request try: - if req.scan_type in ( + # AOB pattern path: req.value is the IDA-style pattern string, + # routed through search_by_pattern. writeable_only doesn't apply + # (pattern scan filters by readability internally; restricting to + # writable-only would silently miss code-section signatures, which + # is the most common AOB use case). + if req.spec.is_pattern: + generator = self._process.search_by_pattern( + req.value, + progress_information=True, + memory_regions=req.memory_regions, + ) + elif req.scan_type in ( ScanTypesEnum.VALUE_BETWEEN, ScanTypesEnum.NOT_VALUE_BETWEEN, ): @@ -140,6 +151,7 @@ def run(self) -> None: self.progress.emit(100.0) self.finished_ok.emit(count) except Exception as exc: # noqa: BLE001 — surface every backend error to the UI + _LOG.warning("First scan failed: %s: %s", type(exc).__name__, exc) self.error.emit(f"{type(exc).__name__}: {exc}") @@ -235,4 +247,5 @@ def run(self) -> None: self.progress.emit(100.0) self.finished_ok.emit(kept) except Exception as exc: # noqa: BLE001 — surface every backend error to the UI + _LOG.warning("Refine scan failed: %s: %s", type(exc).__name__, exc) self.error.emit(f"{type(exc).__name__}: {exc}") diff --git a/PyMemoryEditor/app/scanner_panel.py b/PyMemoryEditor/app/scanner_panel.py index b3c6e2b..fd7d0cf 100644 --- a/PyMemoryEditor/app/scanner_panel.py +++ b/PyMemoryEditor/app/scanner_panel.py @@ -187,21 +187,51 @@ def use_snapshot_cache(self) -> bool: def _refresh_buttons(self) -> None: scanning = self._busy + spec = find_spec(self._type_combo.currentText()) + is_pattern = bool(spec and spec.is_pattern) + self._first_scan_btn.setEnabled(not scanning and not self._has_results) - self._next_scan_btn.setEnabled(not scanning and self._has_results) + # "Next Scan" refines by re-checking the value at each address — that + # concept doesn't apply to a pattern (re-scanning the pattern would + # just re-emit the same addresses), so we hide that path in AOB mode. + self._next_scan_btn.setEnabled( + not scanning and self._has_results and not is_pattern + ) self._update_btn.setEnabled(not scanning and self._has_results) self._new_scan_btn.setEnabled(self._has_results and not scanning) self._cancel_btn.setEnabled(scanning) self._type_combo.setEnabled(not scanning and not self._has_results) - self._scan_combo.setEnabled(not scanning) + # Scan-type combo is *always* disabled in pattern mode (forced to EXACT). + self._scan_combo.setEnabled(not scanning and not is_pattern) self._writable_check.setEnabled(not scanning and not self._has_results) def _on_type_changed(self, label: str) -> None: spec = find_spec(label) if spec is None: return - self._length_spin.setEnabled(spec.accepts_length_override) - if spec.accepts_length_override: + + is_pattern = spec.is_pattern + + # AOB pattern mode reuses the "Value" line for the pattern string and + # hides / forces the rest of the value-shape controls (length, + # second value, scan-type combo) because none of them apply to + # pattern matching. + self._length_spin.setEnabled( + spec.accepts_length_override and not is_pattern + ) + + if is_pattern: + self._value_edit.setPlaceholderText( + 'e.g. "48 8B ? ? 00 00" (IDA-style hex with ? wildcards)' + ) + else: + self._value_edit.setPlaceholderText("e.g. 100 or 0x64 or Hello") + + if is_pattern: + # No meaningful length for an AOB pattern; the scanner derives it. + self._length_spin.setValue(1) + self._length_spin.setSuffix(" bytes") + elif spec.accepts_length_override: if spec.pytype is bytes: self._length_spin.setValue(max(4, self._length_spin.value())) self._length_spin.setSuffix(" bytes") @@ -212,6 +242,23 @@ def _on_type_changed(self, label: str) -> None: self._length_spin.setValue(spec.length) self._length_spin.setSuffix(" bytes") + # Force EXACT_VALUE on pattern mode and disable the scan-type combo + # (Bigger Than / Smaller Than / Between are meaningless for patterns). + if is_pattern: + for index, (_, scan_type) in enumerate(SCAN_TYPE_CHOICES): + if scan_type is ScanTypesEnum.EXACT_VALUE: + self._scan_combo.setCurrentIndex(index) + break + # Ranges + pattern don't mix — make sure the "second value" is + # hidden if a range type was selected before switching to pattern. + self._second_value_edit.hide() + self._second_value_label.hide() + self._scan_combo.setEnabled(not is_pattern and not self._busy) + + # The pattern/non-pattern flag also drives Next-Scan availability, so + # let _refresh_buttons re-evaluate now that the type has flipped. + self._refresh_buttons() + def _on_scan_type_changed(self, index: int) -> None: _, scan_type = SCAN_TYPE_CHOICES[index] ranged = scan_type in ( @@ -228,6 +275,23 @@ def _build_request(self, *, with_value: bool = True) -> Optional[ScanRequest]: _, scan_type = SCAN_TYPE_CHOICES[self._scan_combo.currentIndex()] + # AOB pattern path — value is the pattern string, scan_type is always + # EXACT (the combo was forced + disabled by _on_type_changed), and + # length is irrelevant (the scanner derives it from the pattern). + if spec.is_pattern: + try: + value, length = parse_value(spec, self._value_edit.text()) + except ValueError as exc: + QMessageBox.warning(self, "Invalid pattern", str(exc)) + return None + return ScanRequest( + spec=spec, + length=int(length), + scan_type=ScanTypesEnum.EXACT_VALUE, + value=None if not with_value else value, + writeable_only=self._writable_check.isChecked(), + ) + length_override = ( self._length_spin.value() if spec.accepts_length_override else None ) diff --git a/PyMemoryEditor/app/threads_dialog.py b/PyMemoryEditor/app/threads_dialog.py new file mode 100644 index 0000000..0ea0e88 --- /dev/null +++ b/PyMemoryEditor/app/threads_dialog.py @@ -0,0 +1,232 @@ +# -*- coding: utf-8 -*- +""" +Threads dialog — exposes ``process.get_threads()``. + +Shows every thread the target process currently has, in a sortable table, +with optional auto-refresh. The intent mirrors Cheat Engine's "Process → +Threads" window: you don't typically *act* on threads directly, but seeing +them is useful for introspection (how many workers does this game have? +is the main thread alive?). The optional auto-refresh polls at ~1 Hz so +you can watch threads come and go. + +Lives alongside the existing Memory Map dialog — same shape, same patterns +(background worker, toolbar with Refresh, sortable table, Close button). +""" +from typing import List, Optional + +from PySide6.QtCore import Qt, QThread, QTimer, Signal +from PySide6.QtGui import QStandardItem, QStandardItemModel +from PySide6.QtWidgets import ( + QAbstractItemView, + QCheckBox, + QDialog, + QHBoxLayout, + QHeaderView, + QLabel, + QMessageBox, + QPushButton, + QSpinBox, + QTableView, + QVBoxLayout, +) + +from PyMemoryEditor import AbstractProcess, ThreadInfo + +from ._widgets import NumericItem + + +class _ThreadsWorker(QThread): + """Background thread that runs ``process.get_threads()`` off the UI.""" + + threads_ready = Signal(object) # List[ThreadInfo] + threads_failed = Signal(str) + + def __init__(self, process: AbstractProcess, parent=None): + super().__init__(parent) + self._process = process + + def run(self) -> None: + try: + threads = list(self._process.get_threads()) + except Exception as exc: # noqa: BLE001 + self.threads_failed.emit(str(exc)) + return + self.threads_ready.emit(threads) + + +class ThreadsDialog(QDialog): + """Lists the output of ``get_threads()`` in a sortable table.""" + + def __init__(self, process: AbstractProcess, parent=None): + super().__init__(parent) + self._process = process + self._threads: List[ThreadInfo] = [] + self._worker: Optional[_ThreadsWorker] = None + + self.setWindowTitle(f"Threads — PID {process.pid}") + self.resize(640, 520) + + self._build_ui() + + # Auto-refresh timer; off by default. The interval is matched to the + # main window's heartbeat so the user only sees consistent data even + # if both fire on the same tick. + self._timer = QTimer(self) + self._timer.timeout.connect(self.refresh) + + self.refresh() + + def _build_ui(self) -> None: + layout = QVBoxLayout(self) + layout.setContentsMargins(14, 14, 14, 14) + layout.setSpacing(10) + + header = QLabel( + f"Threads" + f"  PID {self._process.pid}" + ) + header.setTextFormat(Qt.RichText) + layout.addWidget(header) + + self._count_label = QLabel("") + self._count_label.setObjectName("hint") + layout.addWidget(self._count_label) + + bar = QHBoxLayout() + bar.setSpacing(8) + + self._refresh_btn = QPushButton("Refresh") + self._refresh_btn.clicked.connect(self.refresh) + bar.addWidget(self._refresh_btn) + + bar.addStretch(1) + + self._auto_check = QCheckBox("Auto-refresh") + self._auto_check.setToolTip( + "Poll get_threads() at the interval below. Threads die and " + "spawn often — leaving this on lets you watch the churn." + ) + self._auto_check.toggled.connect(self._toggle_auto_refresh) + bar.addWidget(self._auto_check) + + bar.addWidget(QLabel("ms:")) + self._interval_spin = QSpinBox() + self._interval_spin.setRange(200, 10000) + self._interval_spin.setSingleStep(100) + self._interval_spin.setValue(1000) + self._interval_spin.valueChanged.connect(self._sync_timer) + bar.addWidget(self._interval_spin) + + close_btn = QPushButton("Close") + close_btn.clicked.connect(self.accept) + bar.addWidget(close_btn) + layout.addLayout(bar) + + self._model = QStandardItemModel(0, 4, self) + self._model.setHorizontalHeaderLabels( + ["TID", "State", "Priority", "Notes"] + ) + + self._table = QTableView() + self._table.setModel(self._model) + self._table.setSelectionBehavior(QAbstractItemView.SelectRows) + self._table.setSelectionMode(QAbstractItemView.SingleSelection) + self._table.setEditTriggers(QAbstractItemView.NoEditTriggers) + self._table.setSortingEnabled(True) + self._table.setAlternatingRowColors(True) + self._table.verticalHeader().setVisible(False) + self._table.horizontalHeader().setStretchLastSection(True) + self._table.horizontalHeader().setSectionResizeMode( + 0, QHeaderView.ResizeToContents + ) + self._table.horizontalHeader().setSectionResizeMode( + 1, QHeaderView.ResizeToContents + ) + self._table.horizontalHeader().setSectionResizeMode( + 2, QHeaderView.ResizeToContents + ) + layout.addWidget(self._table, 1) + + def refresh(self) -> None: + if self._worker is not None and self._worker.isRunning(): + return + + self._set_busy(True) + self._count_label.setText("Enumerating threads…") + + worker = _ThreadsWorker(self._process, self) + worker.threads_ready.connect(self._on_threads_ready) + worker.threads_failed.connect(self._on_threads_failed) + worker.finished.connect(self._on_worker_finished) + self._worker = worker + worker.start() + + def _set_busy(self, busy: bool) -> None: + self._refresh_btn.setEnabled(not busy) + + def _on_threads_ready(self, threads) -> None: + self._threads = list(threads) + self._model.setRowCount(0) + for info in self._threads: + tid_item = NumericItem(str(info.tid)) + tid_item.setData(int(info.tid), Qt.UserRole) + tid_item.setTextAlignment(Qt.AlignRight | Qt.AlignVCenter) + + state_item = QStandardItem(info.state if info.state is not None else "—") + + priority_text = "—" if info.priority is None else str(info.priority) + priority_item = NumericItem(priority_text) + if info.priority is not None: + priority_item.setData(int(info.priority), Qt.UserRole) + priority_item.setTextAlignment(Qt.AlignRight | Qt.AlignVCenter) + + notes_text = "" + if info.start_address is not None: + notes_text = f"start=0x{info.start_address:X}" + notes_item = QStandardItem(notes_text) + + self._model.appendRow( + [tid_item, state_item, priority_item, notes_item] + ) + + main = min((t.tid for t in self._threads), default=None) + main_str = f" · main TID {main}" if main is not None else "" + self._count_label.setText( + f"{len(self._threads):,} thread(s){main_str}" + ) + + def _on_threads_failed(self, message: str) -> None: + self._count_label.setText("Failed to enumerate threads.") + QMessageBox.critical( + self, "Threads", f"Failed to enumerate threads:\n\n{message}" + ) + + def _on_worker_finished(self) -> None: + self._set_busy(False) + worker = self._worker + self._worker = None + if worker is not None: + worker.deleteLater() + + def _toggle_auto_refresh(self, on: bool) -> None: + if on: + self._sync_timer() + else: + self._timer.stop() + + def _sync_timer(self) -> None: + self._timer.setInterval(int(self._interval_spin.value())) + if self._auto_check.isChecked() and not self._timer.isActive(): + self._timer.start() + + def closeEvent(self, event): # noqa: N802 — Qt naming + self._timer.stop() + if self._worker is not None and self._worker.isRunning(): + try: + self._worker.threads_ready.disconnect() + self._worker.threads_failed.disconnect() + self._worker.finished.disconnect() + except (RuntimeError, TypeError): + pass + self._worker.wait(1000) + super().closeEvent(event) diff --git a/PyMemoryEditor/app/value_types.py b/PyMemoryEditor/app/value_types.py index 6bbff22..ca013b7 100644 --- a/PyMemoryEditor/app/value_types.py +++ b/PyMemoryEditor/app/value_types.py @@ -22,6 +22,12 @@ class ValueTypeSpec: format: Callable[[Any], str] hex_capable: bool = False # Can the value be entered in hex? accepts_length_override: bool = False # True only for str/bytes + # When True the scanner panel routes this type through + # ``process.search_by_pattern`` (AOB / IDA-style hex with wildcards) + # instead of ``search_by_value`` — the "Value" input becomes the pattern + # string and the scan-type / length controls are hidden because they + # don't apply. + is_pattern: bool = False def _parse_bool(text: str) -> bool: @@ -73,6 +79,29 @@ def _parse_bytes(text: str) -> bytes: raise ValueError(f"Invalid byte array: {exc}") +def _parse_pattern(text: str) -> str: + """Validate an IDA-style AOB pattern and return it verbatim. + + The scanner passes the string straight to ``process.search_by_pattern``, + so the parse step is just a "does this compile?" gate that surfaces a + clear ValueError early — much friendlier than letting the scan worker + raise mid-iteration with a low-level message. + """ + from PyMemoryEditor.util.pattern import compile_pattern + + stripped = text.strip() + if not stripped: + raise ValueError( + "Empty pattern. Use IDA syntax: hex bytes separated by spaces, " + "with '?' as a one-byte wildcard. Example: '48 8B ? ? 00'." + ) + # Side-effect: raises ValueError on malformed input. We don't keep the + # compiled regex here — the scanner re-compiles on its end so this is + # purely for early validation feedback. + compile_pattern(stripped) + return stripped + + def _fmt_bytes(value: bytes) -> str: if value is None: return "" @@ -159,6 +188,18 @@ def _fmt_int(value): _fmt_bytes, accepts_length_override=True, ), + # AOB pattern scan — the "Value" input becomes an IDA-style hex string + # with '?' wildcards; the scanner panel hides scan-type / length / "Next + # Scan" because they don't apply. + ValueTypeSpec( + "AOB Pattern (IDA)", + bytes, + 0, + _parse_pattern, + lambda v: "" if v is None else (v if isinstance(v, str) else _fmt_bytes(v)), + accepts_length_override=False, + is_pattern=True, + ), ) @@ -178,6 +219,12 @@ def parse_value( """ value = spec.parse(text) length = spec.length + # AOB patterns short-circuit: ``length`` isn't meaningful — the scanner + # derives the byte width from the pattern itself. Return early so the + # bytes/str length-inference rules below don't accidentally trip on the + # pattern string (whose len() counts characters, not target bytes). + if spec.is_pattern: + return value, 0 if spec.accepts_length_override and length_override is not None: length = max(1, int(length_override)) if spec.pytype is bytes and length_override is None: diff --git a/PyMemoryEditor/linux/functions.py b/PyMemoryEditor/linux/functions.py index 004c8df..149bf69 100644 --- a/PyMemoryEditor/linux/functions.py +++ b/PyMemoryEditor/linux/functions.py @@ -8,22 +8,32 @@ import ctypes import errno as errno_mod +import logging import os from ctypes import addressof, sizeof from typing import Dict, Generator, Optional, Sequence, Tuple, Type, TypeVar, Union from ..enums import ScanTypesEnum from ..process.region import enrich_region -from ..process.scanning import iter_search_results, iter_values_for_addresses +from ..process.scanning import ( + iter_pattern_results, + iter_search_results, + iter_values_for_addresses, +) +from ..process.thread_info import ThreadInfo from ..util import ( _validate_pytype, get_c_type_of, values_to_bytes, ) +from ..util.pattern import PatternLike, compile_pattern from .libc import libc from .types import MEMORY_BASIC_INFORMATION, PATH_SIZE, PRIVILEGES_SIZE, iovec +_logger = logging.getLogger("PyMemoryEditor") + + T = TypeVar("T") @@ -243,6 +253,57 @@ def is_transient(exc: BaseException) -> bool: ) +def search_addresses_by_pattern( + pid: int, + pattern: PatternLike, + *, + byte_length: int = 0, + progress_information: bool = False, + memory_regions: Optional[Sequence[Dict]] = None, +) -> Generator[Union[int, Tuple[int, dict]], None, None]: + """ + AOB scan against every readable, non-shared region of the target. See + :meth:`AbstractProcess.search_by_pattern`. + """ + compiled, length = compile_pattern(pattern, byte_length=byte_length) + + source_regions = ( + memory_regions if memory_regions is not None else get_memory_regions(pid) + ) + + def is_scannable(region) -> bool: + privileges = region["struct"].Privileges + if b"r" not in privileges: + return False + # Mirror search_addresses_by_value: skip shared mappings (libc text + # etc.) — they bloat scans with noise. + if b"s" in privileges: + return False + return True + + filtered_regions = [region for region in source_regions if is_scannable(region)] + filtered_regions.sort(key=lambda region: region["address"]) + + def read_chunk(address: int, size: int): + buffer = (ctypes.c_byte * size)() + _process_vm_readv(pid, addressof(buffer), address, sizeof(buffer)) + return buffer + + def is_transient(exc: BaseException) -> bool: + if isinstance(exc, _LinuxPartialIOError): + return True + return isinstance(exc, OSError) and exc.errno in _PAGE_GONE_ERRNOS + + yield from iter_pattern_results( + filtered_regions, + compiled, + length, + read_chunk, + progress_information=progress_information, + transient_error_check=is_transient, + ) + + def search_values_by_addresses( pid: int, pytype: Type[T], @@ -315,3 +376,63 @@ def write_process_memory( _process_vm_writev(pid, addressof(data), address, sizeof(data)) return value + + +def get_threads(pid: int) -> Generator[ThreadInfo, None, None]: + """ + Yield a :class:`ThreadInfo` for every thread of the target process by + listing ``/proc//task/`` — each subdirectory there is a TID. + + State and priority come from ``/proc//task//stat`` when readable; + silent on permission/race errors (a thread may exit between listing the + directory and reading its stat file) but logged at DEBUG so observers can + see it. + """ + task_dir = "/proc/{}/task".format(pid) + + try: + entries = os.listdir(task_dir) + except OSError as exc: + _logger.debug("get_threads: could not list %s: %s", task_dir, exc) + return + + for entry in entries: + try: + tid = int(entry) + except ValueError: + continue + + state: Optional[str] = None + priority: Optional[int] = None + try: + with open("{}/{}/stat".format(task_dir, entry), "r") as fh: + raw_stat = fh.read() + # /proc//task//stat layout (man 5 proc): + # pid (comm) state ppid pgrp session tty_nr tpgid flags minflt + # cminflt majflt cmajflt utime stime cutime cstime priority ... + # ``comm`` is wrapped in parens and *may itself contain whitespace + # or parentheses*, so the only safe split point is the last ')'. + close_paren = raw_stat.rfind(")") + if close_paren != -1: + rest = raw_stat[close_paren + 1 :].split() + # rest[0] = state, rest[15] = priority (field 18 in the man + # page, with the first two fields already consumed). + if rest: + state = rest[0] + if len(rest) > 15: + try: + priority = int(rest[15]) + except ValueError: + priority = None + except OSError as exc: + _logger.debug( + "get_threads: could not read stat for tid=%s: %s", entry, exc + ) + + yield ThreadInfo( + tid=tid, + start_address=None, + state=state, + priority=priority, + raw=entry, + ) diff --git a/PyMemoryEditor/linux/process.py b/PyMemoryEditor/linux/process.py index 8a87f1c..1ac11cb 100644 --- a/PyMemoryEditor/linux/process.py +++ b/PyMemoryEditor/linux/process.py @@ -7,9 +7,12 @@ from ..process import AbstractProcess from ..process.errors import ClosedProcess from ..util import resolve_bufflength +from ..process.thread_info import ThreadInfo from .functions import ( get_memory_regions, + get_threads, read_process_memory, + search_addresses_by_pattern, search_addresses_by_value, search_values_by_addresses, write_process_memory, @@ -31,6 +34,7 @@ def __init__( pid: Optional[int] = None, permission=None, case_sensitive: bool = True, + exact_match: bool = True, ): """ :param process_name: name of the target process. @@ -41,11 +45,14 @@ def __init__( mask doesn't disappear silently here — pass ``None`` (or omit) on non-Windows platforms. :param case_sensitive: when False, process_name matching ignores case. + :param exact_match: when False, ``process_name`` is matched as a + substring (e.g. ``"chrome"`` finds ``"chromium-browser"``). """ super().__init__( process_name=process_name, pid=pid, case_sensitive=case_sensitive, + exact_match=exact_match, ) self.__closed = False @@ -73,6 +80,10 @@ def get_memory_regions(self) -> Generator[dict, None, None]: self.__require_open() return get_memory_regions(self.pid) + def get_threads(self) -> Generator[ThreadInfo, None, None]: + self.__require_open() + return get_threads(self.pid) + def read_process_memory( self, address: int, @@ -132,6 +143,23 @@ def search_by_value( memory_regions=memory_regions, ) + def search_by_pattern( + self, + pattern, + *, + byte_length: int = 0, + progress_information: bool = False, + memory_regions: Optional[Sequence[Dict]] = None, + ) -> Generator[Union[int, Tuple[int, dict]], None, None]: + self.__require_open() + return search_addresses_by_pattern( + self.pid, + pattern, + byte_length=byte_length, + progress_information=progress_information, + memory_regions=memory_regions, + ) + def search_by_value_between( self, pytype: Type[T], diff --git a/PyMemoryEditor/macos/functions.py b/PyMemoryEditor/macos/functions.py index 0f4b91d..6c5b08f 100644 --- a/PyMemoryEditor/macos/functions.py +++ b/PyMemoryEditor/macos/functions.py @@ -6,18 +6,25 @@ """ import ctypes +import logging import os import warnings from typing import Dict, Generator, Optional, Sequence, Tuple, Type, TypeVar, Union from ..enums import ScanTypesEnum from ..process.region import enrich_region -from ..process.scanning import iter_search_results, iter_values_for_addresses +from ..process.scanning import ( + iter_pattern_results, + iter_search_results, + iter_values_for_addresses, +) +from ..process.thread_info import ThreadInfo from ..util import ( _validate_pytype, get_c_type_of, values_to_bytes, ) +from ..util.pattern import PatternLike, compile_pattern from .libsystem import libsystem, mach_error_message, mach_task_self_ from .types import ( @@ -47,6 +54,9 @@ _WRITE_RETRY_CODES = (KERN_PROTECTION_FAILURE, KERN_INVALID_ADDRESS) +_logger = logging.getLogger("PyMemoryEditor") + + T = TypeVar("T") @@ -251,7 +261,7 @@ def _mach_write(task: int, address: int, local_buffer_address: int, size: int) - task, address, size, 0, original_protection ) if restore_kr != KERN_SUCCESS: - warnings.warn( + message = ( "mach_vm_protect could not restore the original protection " "(0x%x) on the target page at 0x%x after a write-via-protect-flip; " "the page is left more permissive than before (kr=%d, %s)." @@ -260,10 +270,10 @@ def _mach_write(task: int, address: int, local_buffer_address: int, size: int) - address, restore_kr, mach_error_message(restore_kr), - ), - ResourceWarning, - stacklevel=2, + ) ) + _logger.warning(message) + warnings.warn(message, ResourceWarning, stacklevel=2) def _query_region(task: int, address: int): @@ -402,6 +412,91 @@ def is_transient(exc: BaseException) -> bool: ) +def get_threads(task: int) -> Generator[ThreadInfo, None, None]: + """ + Yield a :class:`ThreadInfo` for every thread of the target task using + Mach's ``task_threads``. + + .. note:: + ``tid`` here is the **Mach thread port name**, not the BSD/POSIX + pthread id. Looking up the POSIX tid would require an extra + ``thread_info(THREAD_IDENTIFIER_INFO)`` call per thread; the Mach port + is sufficient for any further Mach-level operation and is what the + kernel hands us cheaply. + """ + thread_list = ctypes.POINTER(ctypes.c_uint)() + count = ctypes.c_uint(0) + + kr = libsystem.task_threads(task, ctypes.byref(thread_list), ctypes.byref(count)) + if kr != KERN_SUCCESS: + raise OSError( + "task_threads failed: %s (kr=%d)" % (mach_error_message(kr), kr) + ) + + try: + for index in range(count.value): + yield ThreadInfo( + tid=int(thread_list[index]), + start_address=None, + state=None, + priority=None, + raw=int(thread_list[index]), + ) + finally: + # The kernel out-allocates ``thread_list`` in the caller's address + # space; freeing it back is the caller's responsibility, otherwise we + # leak VM in *our own* task each enumeration. The deallocation size is + # ``count * sizeof(mach_port_t)``. + if count.value: + libsystem.vm_deallocate( + mach_task_self_.value, + ctypes.cast(thread_list, ctypes.c_void_p), + count.value * ctypes.sizeof(ctypes.c_uint), + ) + + +def search_addresses_by_pattern( + task: int, + pattern: PatternLike, + *, + byte_length: int = 0, + progress_information: bool = False, + memory_regions: Optional[Sequence[Dict]] = None, +) -> Generator[Union[int, Tuple[int, dict]], None, None]: + """ + AOB scan against every readable region of the target task. See + :meth:`AbstractProcess.search_by_pattern`. + """ + compiled, length = compile_pattern(pattern, byte_length=byte_length) + + source_regions = ( + memory_regions if memory_regions is not None else get_memory_regions(task) + ) + + def is_scannable(region) -> bool: + return (region["struct"].Protection & VM_PROT_READ) != 0 + + filtered_regions = [region for region in source_regions if is_scannable(region)] + filtered_regions.sort(key=lambda region: region["address"]) + + def read_chunk(address: int, size: int): + buffer = (ctypes.c_byte * size)() + _mach_read(task, address, ctypes.addressof(buffer), size) + return buffer + + def is_transient(exc: BaseException) -> bool: + return isinstance(exc, MachReadError) and exc.kr in _PAGE_GONE_KRS + + yield from iter_pattern_results( + filtered_regions, + compiled, + length, + read_chunk, + progress_information=progress_information, + transient_error_check=is_transient, + ) + + def search_values_by_addresses( task: int, pytype: Type[T], diff --git a/PyMemoryEditor/macos/libsystem.py b/PyMemoryEditor/macos/libsystem.py index 51d32f4..edafebc 100644 --- a/PyMemoryEditor/macos/libsystem.py +++ b/PyMemoryEditor/macos/libsystem.py @@ -107,6 +107,29 @@ libsystem.mach_port_deallocate.argtypes = (mach_port_t, mach_port_t) libsystem.mach_port_deallocate.restype = kern_return_t +# kern_return_t task_threads( +# task_inspect_t target_task, +# thread_act_array_t *act_list, /* out: kernel-allocated array of thread ports */ +# mach_msg_type_number_t *act_listCnt); +libsystem.task_threads.argtypes = ( + task_t, + POINTER(POINTER(mach_port_t)), + POINTER(mach_msg_type_number_t), +) +libsystem.task_threads.restype = kern_return_t + +# kern_return_t vm_deallocate( +# vm_map_t target_task, +# vm_address_t address, +# vm_size_t size); +# Used to free the thread_act_array_t returned by task_threads. +libsystem.vm_deallocate.argtypes = ( + vm_map_t, + ctypes.c_void_p, + ctypes.c_size_t, +) +libsystem.vm_deallocate.restype = kern_return_t + # struct rusage_info_v0 — first slice of rusage_info_t. ri_phys_footprint is # the number Activity Monitor's "Memory" column shows (anonymous + compressed diff --git a/PyMemoryEditor/macos/process.py b/PyMemoryEditor/macos/process.py index 9a52707..1df0ec6 100644 --- a/PyMemoryEditor/macos/process.py +++ b/PyMemoryEditor/macos/process.py @@ -6,13 +6,16 @@ from ..enums import ScanTypesEnum from ..process import AbstractProcess from ..process.errors import ClosedProcess +from ..process.thread_info import ThreadInfo from ..util import resolve_bufflength from .functions import ( get_memory_regions, get_task_for_pid, + get_threads, read_process_memory, release_task, + search_addresses_by_pattern, search_addresses_by_value, search_values_by_addresses, write_process_memory, @@ -39,6 +42,7 @@ def __init__( pid: Optional[int] = None, permission=None, case_sensitive: bool = True, + exact_match: bool = True, ): """ :param process_name: name of the target process. @@ -49,11 +53,14 @@ def __init__( mask doesn't disappear silently here — pass ``None`` (or omit) on non-Windows platforms. :param case_sensitive: when False, process_name matching ignores case. + :param exact_match: when False, ``process_name`` is matched as a + substring (e.g. ``"chrome"`` finds ``"Google Chrome"``). """ super().__init__( process_name=process_name, pid=pid, case_sensitive=case_sensitive, + exact_match=exact_match, ) # `permission` is accepted for cross-platform parity but has no effect @@ -112,6 +119,10 @@ def get_memory_regions(self) -> Generator[dict, None, None]: self.__require_open() return get_memory_regions(self.__task) + def get_threads(self) -> Generator[ThreadInfo, None, None]: + self.__require_open() + return get_threads(self.__task) + def search_by_addresses( self, pytype: Type[T], @@ -160,6 +171,23 @@ def search_by_value( memory_regions=memory_regions, ) + def search_by_pattern( + self, + pattern, + *, + byte_length: int = 0, + progress_information: bool = False, + memory_regions: Optional[Sequence[Dict]] = None, + ) -> Generator[Union[int, Tuple[int, dict]], None, None]: + self.__require_open() + return search_addresses_by_pattern( + self.__task, + pattern, + byte_length=byte_length, + progress_information=progress_information, + memory_regions=memory_regions, + ) + def search_by_value_between( self, pytype: Type[T], diff --git a/PyMemoryEditor/process/abstract.py b/PyMemoryEditor/process/abstract.py index 5f7d356..d56fb61 100644 --- a/PyMemoryEditor/process/abstract.py +++ b/PyMemoryEditor/process/abstract.py @@ -1,4 +1,5 @@ # -*- coding: utf-8 -*- +import sys from abc import ABC, abstractmethod from typing import ( Dict, @@ -15,6 +16,7 @@ from ..enums import ScanTypesEnum from .info import ProcessInfo from .scanning import _PRESORTED_KEY +from .thread_info import ThreadInfo T = TypeVar("T") @@ -32,12 +34,17 @@ def __init__( process_name: Optional[str] = None, pid: Optional[int] = None, case_sensitive: bool = True, + exact_match: bool = True, ): """ :param process_name: name of the target process. :param pid: process ID. :param case_sensitive: when False, process_name matching ignores case (recommended on Windows where process names are case-insensitive). + :param exact_match: when False, ``process_name`` is matched as a + substring — ``"chrome"`` matches ``"chrome.exe"`` / ``"Google Chrome"``. + If more than one process matches, ``AmbiguousProcessNameError`` is + raised so you can pick a PID from the list. """ self._process_info = ProcessInfo() @@ -46,7 +53,9 @@ def __init__( elif process_name: self._process_info.set_process_name( - process_name, case_sensitive=case_sensitive + process_name, + case_sensitive=case_sensitive, + exact_match=exact_match, ) else: @@ -79,6 +88,36 @@ def get_memory_regions(self) -> Generator[dict, None, None]: """ raise NotImplementedError() + @abstractmethod + def get_threads(self) -> Generator[ThreadInfo, None, None]: + """ + Yield a :class:`~PyMemoryEditor.ThreadInfo` for every thread running + inside the target process. + + The fields that each backend can fill in cheaply vary — see + ``ThreadInfo`` for which attributes may be ``None`` per platform. + The ``tid`` field's *meaning* is platform-specific (POSIX TID on + Linux, DWORD TID on Windows, Mach port name on macOS). + + Use :attr:`main_thread` for the conventional "main thread" shortcut. + """ + raise NotImplementedError() + + @property + def main_thread(self) -> Optional[ThreadInfo]: + """ + The conventional "main thread" of the target — by convention, the + thread with the smallest ``tid``. Returns ``None`` if the target has + no listable threads (rare; typically means the process just exited). + + Useful as a quick hand-off into thread-specific operations, and as a + sanity check ("is anything still running in there?"). + """ + threads = list(self.get_threads()) + if not threads: + return None + return min(threads, key=lambda t: t.tid) + def snapshot_memory_regions(self) -> List[Dict]: """ Return a materialized snapshot of the process memory regions. @@ -149,6 +188,34 @@ def search_by_value( """ raise NotImplementedError() + @abstractmethod + def search_by_pattern( + self, + pattern: Union[str, bytes, "object"], + *, + byte_length: int = 0, + progress_information: bool = False, + memory_regions: Optional[Sequence[Dict]] = None, + ) -> Generator[Union[int, Tuple[int, dict]], None, None]: + """ + Scan the target's memory for a byte pattern (AOB) — the Cheat Engine / + IDA technique for locating code or data that moves between builds. + + :param pattern: one of the forms accepted by + :func:`PyMemoryEditor.util.pattern.compile_pattern` — an IDA-style + hex string with ``?`` wildcards (``"48 8B ? ? 00"``), a raw bytes + regex, or a pre-compiled ``re.Pattern[bytes]``. + :param byte_length: required when ``pattern`` is a regex / pre-compiled + Pattern — the number of bytes one match consumes. Ignored for + IDA-style strings (inferred from the token count). + :param progress_information: if True, yields ``(address, info)`` + tuples (same shape as ``search_by_value``). + :param memory_regions: optional snapshot from + ``snapshot_memory_regions()`` to skip region enumeration on + iterative workflows. + """ + raise NotImplementedError() + @abstractmethod def search_by_value_between( self, @@ -215,3 +282,65 @@ def write_process_memory( :param value: value to be written. """ raise NotImplementedError() + + def resolve_pointer_chain( + self, + base_address: int, + offsets: Sequence[int], + *, + ptr_size: int = 8, + ) -> int: + """ + Walk a multi-level pointer chain — the kind of recipe Cheat Engine + exports for addresses that survive a process restart. + + Reads ``ptr_size`` bytes at ``base_address`` to obtain the first + pointer, then for each offset in ``offsets[:-1]`` adds the offset and + dereferences again. The **last** offset is added *without* + dereferencing — the returned integer is the final address where the + value of interest lives. Read or write it with the regular + ``read_process_memory`` / ``write_process_memory`` calls. + + :param base_address: starting address — typically + ``module_base + static_offset``. + :param offsets: sequence of offsets to walk. Pass ``[]`` to dereference + ``base_address`` once and return that pointer. + :param ptr_size: pointer width — 8 for 64-bit targets (default), 4 for + 32-bit targets. + + Example + ------- + Cheat-Engine cheat table entry:: + + "game.exe" + 0x10F4F4 -> [+0x0] -> [+0x158] ; HP + + Translates to:: + + hp_addr = process.resolve_pointer_chain(0x14010F4F4, [0x0, 0x158]) + hp = process.read_process_memory(hp_addr, int, 4) + """ + if ptr_size not in (4, 8): + raise ValueError( + "ptr_size must be 4 (32-bit target) or 8 (64-bit target)." + ) + + # ``read_process_memory(.., int, ..)`` decodes as a *signed* integer + # (see util.convert.get_c_type_of). Pointers in the upper half of the + # address space would come back negative and the next dereference would + # land at an invalid kernel-side address. Read as raw bytes and + # reinterpret as unsigned so every pointer fits the OS's natural range. + byte_order = sys.byteorder + + def _read_ptr(addr: int) -> int: + raw = self.read_process_memory(addr, bytes, ptr_size) + return int.from_bytes(raw, byte_order, signed=False) + + if not offsets: + return _read_ptr(base_address) + + current = _read_ptr(base_address) + + for offset in offsets[:-1]: + current = _read_ptr(current + offset) + + return current + offsets[-1] diff --git a/PyMemoryEditor/process/info.py b/PyMemoryEditor/process/info.py index 3df0d78..b70cdc9 100644 --- a/PyMemoryEditor/process/info.py +++ b/PyMemoryEditor/process/info.py @@ -39,10 +39,16 @@ def process_name(self, process_name: str) -> None: self.set_process_name(process_name) def set_process_name( - self, process_name: str, *, case_sensitive: bool = True + self, + process_name: str, + *, + case_sensitive: bool = True, + exact_match: bool = True, ) -> None: pid = get_process_id_by_process_name( - process_name, case_sensitive=case_sensitive + process_name, + case_sensitive=case_sensitive, + exact_match=exact_match, ) if pid is None: raise ProcessNotFoundError(process_name) diff --git a/PyMemoryEditor/process/scanning.py b/PyMemoryEditor/process/scanning.py index 470188f..14f07b9 100644 --- a/PyMemoryEditor/process/scanning.py +++ b/PyMemoryEditor/process/scanning.py @@ -22,6 +22,7 @@ """ import ctypes +import logging from typing import ( Any, Callable, @@ -29,6 +30,7 @@ Generator, Iterable, Optional, + Pattern, Sequence, Tuple, Type, @@ -46,6 +48,9 @@ ) +_logger = logging.getLogger("PyMemoryEditor") + + # Shared type for the in-region search callable. ``scan_memory`` accepts a # tuple target (for VALUE_BETWEEN) while ``scan_memory_for_exact_value`` does # not — at runtime we only ever route VALUE_BETWEEN through ``scan_memory``, @@ -176,6 +181,12 @@ def iter_values_for_addresses( transient = transient_error_check(exc) if not transient and raise_error: raise + _logger.debug( + "iter_values_for_addresses: skipping chunk at 0x%X (%d bytes): %s", + chunk_address, + read_size, + exc, + ) while ( address_index < len(sorted_addresses) and sorted_addresses[address_index] < chunk_end @@ -297,6 +308,12 @@ def iter_search_results( chunk_data = read_chunk(chunk_address, read_size) except Exception as exc: # noqa: BLE001 — backend errors vary if transient_error_check(exc): + _logger.debug( + "iter_search_results: skipping chunk at 0x%X (%d bytes): %s", + chunk_address, + read_size, + exc, + ) continue raise @@ -337,4 +354,104 @@ def iter_search_results( checked_memory_size += size -__all__ = ("iter_search_results", "iter_values_for_addresses") +def iter_pattern_results( + memory_regions: Sequence[Dict], + compiled_pattern: "Pattern[bytes]", + pattern_length: int, + read_chunk: Callable[[int, int], Any], + *, + progress_information: bool = False, + transient_error_check: Optional[Callable[[BaseException], bool]] = None, +) -> Generator[Union[int, Tuple[int, dict]], None, None]: + """ + Walk every chunk of every region and yield the addresses where + ``compiled_pattern`` matches. Implements AOB (Array Of Bytes) scanning, + the technique used by Cheat Engine / IDA to locate code or data that + moves between builds. + + Mirrors :func:`iter_search_results` for the regex case: same chunking + strategy, same transient-error classification, same optional + ``progress_information`` shape. ``pattern_length`` is the number of + bytes each match consumes — needed because the regex source length is + not a reliable proxy (an IDA-style ``"48 ? ? 00"`` matches 4 bytes but + its escaped source is longer). The chunk overlap reads + ``pattern_length - 1`` extra bytes from the next chunk so matches + straddling a chunk boundary are still detected; matches that fall in + the overlap region are clamped so each address is attributed to + exactly one chunk. + """ + if transient_error_check is None: + transient_error_check = _always_false + + pattern_length = max(1, pattern_length) + + memory_total = 0 + for region in memory_regions: + memory_total += region["size"] + + if memory_total == 0: + return + + checked_memory_size = 0 + + for region in memory_regions: + address, size = region["address"], region["size"] + + for chunk_offset, chunk_size in iter_region_chunks(size, pattern_length): + chunk_address = address + chunk_offset + + is_last_chunk = chunk_offset + chunk_size >= size + overlap = 0 if is_last_chunk else pattern_length - 1 + read_size = chunk_size + overlap + + try: + chunk_data = read_chunk(chunk_address, read_size) + except Exception as exc: # noqa: BLE001 — backend errors vary + if transient_error_check(exc): + _logger.debug( + "iter_pattern_results: skipping chunk at 0x%X (%d bytes): %s", + chunk_address, + read_size, + exc, + ) + continue + raise + + if chunk_data is None: + continue + + # ``compiled_pattern.finditer`` needs a real bytes object. Convert + # once per chunk; ctypes arrays expose the buffer protocol so this + # is a single memcpy. + data_bytes = bytes(chunk_data) + + for match in compiled_pattern.finditer(data_bytes): + offset = match.start() + # Same clamping rule as iter_search_results: matches in the + # overlap belong to the *next* chunk's emission slot. + if offset >= chunk_size: + continue + found_address = chunk_address + offset + + if progress_information: + yield ( + found_address, + { + "memory_total": memory_total, + "progress": ( + checked_memory_size + chunk_offset + offset + ) + / memory_total, + }, + ) + else: + yield found_address + + checked_memory_size += size + + +__all__ = ( + "iter_pattern_results", + "iter_search_results", + "iter_values_for_addresses", +) diff --git a/PyMemoryEditor/process/thread_info.py b/PyMemoryEditor/process/thread_info.py new file mode 100644 index 0000000..48983da --- /dev/null +++ b/PyMemoryEditor/process/thread_info.py @@ -0,0 +1,51 @@ +# -*- coding: utf-8 -*- + +""" +Cross-platform thread descriptor returned by ``AbstractProcess.get_threads()``. + +Each backend fills in what its OS exposes cheaply; fields left ``None`` mean +"this platform does not surface that attribute via the API we use." Callers +that need a platform-specific extra (e.g. the TEB on Windows, the Mach +thread port on macOS) can pull it through ``raw`` — the backend stores the +original platform handle there for round-tripping. + +The meaning of ``tid`` is intentionally not unified across OSes: + +- **Linux**: POSIX TID — same namespace as PID; ``gettid()`` returns this. +- **Windows**: kernel-assigned global thread id (DWORD) from ``THREADENTRY32``. +- **macOS**: Mach thread port name from ``task_threads``. Not the BSD pthread + id; obtain that via ``thread_info(THREAD_IDENTIFIER_INFO)`` if needed. + +Documented this way because pretending otherwise leads to subtle bugs in code +that mixes pids and tids across platforms. +""" + +from dataclasses import dataclass, field +from typing import Any, Optional + + +@dataclass(frozen=True) +class ThreadInfo: + """A single thread inside a target process. + + :param tid: thread identifier (see module docstring — meaning is platform-dependent). + :param start_address: entry point of the thread, when the OS exposes it + cheaply. ``None`` when not available. + :param state: short human-readable state — e.g. ``"R"`` / ``"S"`` on Linux. + ``None`` when not available. + :param priority: scheduling priority value as reported by the OS. The scale + is platform-specific; ``None`` when not available. + :param raw: the underlying platform handle/struct used to look up this + thread (a ``THREADENTRY32`` on Windows, the TID string from + ``/proc//task/`` on Linux, a Mach port int on macOS). Useful for + advanced callers that need to make follow-up OS-specific calls. + """ + + tid: int + start_address: Optional[int] = None + state: Optional[str] = None + priority: Optional[int] = None + raw: Any = field(default=None, compare=False, repr=False) + + +__all__ = ("ThreadInfo",) diff --git a/PyMemoryEditor/process/util.py b/PyMemoryEditor/process/util.py index 440296c..aa4983a 100644 --- a/PyMemoryEditor/process/util.py +++ b/PyMemoryEditor/process/util.py @@ -8,13 +8,20 @@ def get_process_ids_by_process_name( - process_name: str, *, case_sensitive: bool = True + process_name: str, + *, + case_sensitive: bool = True, + exact_match: bool = True, ) -> List[int]: """ Return a list of all process IDs matching the provided name. :param process_name: process name to search. :param case_sensitive: when False, comparison ignores case (useful on Windows). + :param exact_match: when False, returns every process whose name *contains* + ``process_name`` as a substring — handy when you don't know the exact + executable name (``"chrome"`` matches ``"chrome.exe"``, ``"Google Chrome"``, + ``"Chromium"``, ...). Often combined with ``case_sensitive=False``. """ if not case_sensitive: process_name_cmp = process_name.casefold() @@ -29,14 +36,24 @@ def get_process_ids_by_process_name( except (psutil.NoSuchProcess, psutil.AccessDenied): continue - if (name if case_sensitive else name.casefold()) == process_name_cmp: + name_cmp = name if case_sensitive else name.casefold() + + if exact_match: + hit = name_cmp == process_name_cmp + else: + hit = process_name_cmp in name_cmp + + if hit: matches.append(process.info["pid"]) return matches def get_process_id_by_process_name( - process_name: str, *, case_sensitive: bool = True + process_name: str, + *, + case_sensitive: bool = True, + exact_match: bool = True, ) -> Optional[int]: """ Return the PID of the process matching the provided name. @@ -45,7 +62,9 @@ def get_process_id_by_process_name( Returns None when no process matches (callers should handle this). """ matches = get_process_ids_by_process_name( - process_name, case_sensitive=case_sensitive + process_name, + case_sensitive=case_sensitive, + exact_match=exact_match, ) if len(matches) > 1: diff --git a/PyMemoryEditor/util/__init__.py b/PyMemoryEditor/util/__init__.py index 0e9fed8..37cd207 100644 --- a/PyMemoryEditor/util/__init__.py +++ b/PyMemoryEditor/util/__init__.py @@ -8,6 +8,7 @@ value_to_bytes, values_to_bytes, ) +from .pattern import PatternLike, compile_pattern from .scan import ( DEFAULT_MAX_REGION_CHUNK, iter_region_chunks, diff --git a/PyMemoryEditor/util/pattern.py b/PyMemoryEditor/util/pattern.py new file mode 100644 index 0000000..95d9b17 --- /dev/null +++ b/PyMemoryEditor/util/pattern.py @@ -0,0 +1,114 @@ +# -*- coding: utf-8 -*- + +""" +Pattern compilation for "Array Of Bytes" (AOB) scanning — the technique used +by Cheat Engine, IDA, and most game-hacking tools to locate code/data that +shifts between builds. + +Two input shapes are accepted: + +1. **IDA-style** string with ``?`` / ``??`` wildcards:: + + compile_pattern("48 8B ? ? 00 00") + + Every space-separated token must be either two hex digits (``"48"``) or a + single/double ``?`` for a one-byte wildcard. Whitespace between tokens is + free-form. This is the format almost every public AOB recipe online uses. + +2. **Raw regex bytes** — passed through unchanged with ``re.DOTALL`` so that + the regex meta character ``.`` is allowed to match any byte (including + newlines, which is what you want when scanning binary memory):: + + compile_pattern(rb"\\x48\\x8B..\\x00\\x00") + +The function returns a compiled ``re.Pattern[bytes]`` ready to be used with +``finditer`` against memory chunks read from the target. +""" + +import re +from typing import Pattern, Tuple, Union + + +PatternLike = Union[str, bytes, "re.Pattern[bytes]"] + + +def compile_pattern( + pattern: PatternLike, + *, + byte_length: int = 0, +) -> Tuple["Pattern[bytes]", int]: + """Compile ``pattern`` into a bytes ``re.Pattern`` plus the **number of + bytes** each successful match consumes. + + Returns a ``(compiled_regex, byte_length)`` pair. The second value is the + width of one match in the target's memory — the scanner uses it to compute + chunk-overlap so a match straddling a chunk boundary still gets emitted. + + :param pattern: one of: + + * An **IDA-style hex string** with ``?`` / ``??`` wildcards + (``"48 8B ? ? 00"``) — most cheat-table dumps use this format. Each + token is one byte; the returned ``byte_length`` equals the number of + tokens. + * A **raw bytes regex** (``rb"\\x48\\x8B..\\x00"``). Compiled with + ``re.DOTALL`` so ``.`` matches any byte. **You must pass** + ``byte_length=`` for this form — there is no general way to infer + how many bytes a regex consumes. + * An **already-compiled** ``re.Pattern[bytes]``: same rule — you must + pass ``byte_length=``. + + :raises ValueError: malformed IDA-style token, or ``byte_length`` omitted + for a regex / pre-compiled pattern. + """ + if isinstance(pattern, re.Pattern): + if byte_length <= 0: + raise ValueError( + "byte_length= is required when passing a pre-compiled regex " + "(its source length is not the same as the matched length)." + ) + return pattern, byte_length + + if isinstance(pattern, bytes): + if byte_length <= 0: + raise ValueError( + "byte_length= is required when passing a raw bytes regex " + "(its source length is not the same as the matched length)." + ) + return re.compile(pattern, re.DOTALL), byte_length + + if not isinstance(pattern, str): + raise TypeError( + "Pattern must be str (IDA-style), bytes (regex) or a " + "compiled re.Pattern, not %r" % type(pattern).__name__ + ) + + tokens = pattern.split() + if not tokens: + raise ValueError("Empty pattern.") + + parts = [] + for token in tokens: + if token in ("?", "??"): + # Single-byte wildcard. ``.`` together with re.DOTALL matches any + # byte 0x00-0xFF without special-casing 0x0A. + parts.append(b".") + continue + if len(token) != 2: + raise ValueError( + "Pattern token %r is not two hex digits or a '?' wildcard. " + "Example of a valid pattern: '48 8B ? ? 00'." % token + ) + try: + byte = bytes.fromhex(token) + except ValueError as exc: + raise ValueError( + "Pattern token %r is not valid hex: %s" % (token, exc) + ) + # Escape the byte so e.g. 0x5C (backslash) or 0x28 ('(') don't get + # interpreted as regex meta chars. + parts.append(re.escape(byte)) + + return re.compile(b"".join(parts), re.DOTALL), len(tokens) + + +__all__ = ("compile_pattern", "PatternLike") diff --git a/PyMemoryEditor/win32/functions.py b/PyMemoryEditor/win32/functions.py index ea477f4..9bafe96 100644 --- a/PyMemoryEditor/win32/functions.py +++ b/PyMemoryEditor/win32/functions.py @@ -8,16 +8,23 @@ import ctypes import ctypes.wintypes +import logging from typing import Dict, Generator, Optional, Sequence, Tuple, Type, TypeVar, Union from ..enums import ScanTypesEnum from ..process.region import enrich_region -from ..process.scanning import iter_search_results, iter_values_for_addresses +from ..process.scanning import ( + iter_pattern_results, + iter_search_results, + iter_values_for_addresses, +) +from ..process.thread_info import ThreadInfo from ..util import ( _validate_pytype, get_c_type_of, values_to_bytes, ) +from ..util.pattern import PatternLike, compile_pattern from .enums import MemoryAllocationStatesEnum, MemoryProtectionsEnum, MemoryTypesEnum from .types import ( @@ -25,9 +32,14 @@ MEMORY_BASIC_INFORMATION_32, MEMORY_BASIC_INFORMATION_64, SYSTEM_INFO, + TH32CS_SNAPTHREAD, + THREADENTRY32, ) +_logger = logging.getLogger("PyMemoryEditor") + + # Load the libraries with `use_last_error=True` so that `ctypes.get_last_error()` # returns the per-call `GetLastError` set by the Win32 API. The default # `ctypes.windll.kernel32` accessor uses the shared `WinError` state and @@ -89,6 +101,27 @@ ) kernel32.IsWow64Process.restype = ctypes.wintypes.BOOL +# HANDLE CreateToolhelp32Snapshot(DWORD dwFlags, DWORD th32ProcessID); +# Snapshot of system threads (with TH32CS_SNAPTHREAD); per the docs, the +# ProcessID arg is ignored when SNAPTHREAD is set — the snapshot is global. +kernel32.CreateToolhelp32Snapshot.argtypes = ( + ctypes.wintypes.DWORD, + ctypes.wintypes.DWORD, +) +kernel32.CreateToolhelp32Snapshot.restype = ctypes.wintypes.HANDLE + +kernel32.Thread32First.argtypes = ( + ctypes.wintypes.HANDLE, + ctypes.POINTER(THREADENTRY32), +) +kernel32.Thread32First.restype = ctypes.wintypes.BOOL + +kernel32.Thread32Next.argtypes = ( + ctypes.wintypes.HANDLE, + ctypes.POINTER(THREADENTRY32), +) +kernel32.Thread32Next.restype = ctypes.wintypes.BOOL + system_information = SYSTEM_INFO() kernel32.GetSystemInfo(ctypes.byref(system_information)) @@ -332,6 +365,47 @@ def read_chunk(address: int, size: int): ) +def SearchAddressesByPattern( + process_handle: int, + pattern: PatternLike, + *, + byte_length: int = 0, + progress_information: bool = False, + memory_regions: Optional[Sequence[Dict]] = None, +) -> Generator[Union[int, Tuple[int, dict]], None, None]: + """ + AOB scan against every scannable region of the target process. See + :meth:`AbstractProcess.search_by_pattern`. + """ + compiled, length = compile_pattern(pattern, byte_length=byte_length) + + source_regions = ( + memory_regions + if memory_regions is not None + else GetMemoryRegions(process_handle) + ) + filtered_regions = [ + region + for region in source_regions + if _is_region_scannable(region, writeable_only=False) + ] + filtered_regions.sort(key=lambda region: region["address"]) + + def read_chunk(address: int, size: int): + # ``_read_region`` returns None on transient failures (page unmapped / + # made inaccessible mid-scan); the helper accepts None directly and + # skips the chunk — no exception classification needed here. + return _read_region(process_handle, address, size) + + yield from iter_pattern_results( + filtered_regions, + compiled, + length, + read_chunk, + progress_information=progress_information, + ) + + class _Win32ChunkReadError(OSError): """Raised internally when ReadProcessMemory returns 0 during chunked reads.""" @@ -397,6 +471,52 @@ def is_transient(exc: BaseException) -> bool: ) +def GetThreads(pid: int) -> Generator[ThreadInfo, None, None]: + """ + Yield a :class:`ThreadInfo` for every thread of the target process. + + Uses ``CreateToolhelp32Snapshot(TH32CS_SNAPTHREAD)`` followed by + Thread32First/Next; this is the documented user-mode way to enumerate + threads on Windows without an extra dependency. Caller does not need a + process handle (and therefore no PROCESS_* permission) — the snapshot is + system-wide and we filter by ``th32OwnerProcessID``. + """ + snapshot = kernel32.CreateToolhelp32Snapshot(TH32CS_SNAPTHREAD, 0) + # Per the docs the function returns INVALID_HANDLE_VALUE (-1 cast to HANDLE) + # on failure; in ctypes that comes back as a falsy value once we read it. + if not snapshot or snapshot == ctypes.wintypes.HANDLE(-1).value: + _raise_last_error("CreateToolhelp32Snapshot") + + entry = THREADENTRY32() + entry.dwSize = ctypes.sizeof(entry) + + try: + if not kernel32.Thread32First(snapshot, ctypes.byref(entry)): + # Empty snapshot is legal (no threads visible). Log and bail. + _logger.debug( + "GetThreads: Thread32First returned 0 (snapshot empty for pid=%d)", + pid, + ) + return + + while True: + if entry.th32OwnerProcessID == pid: + yield ThreadInfo( + tid=entry.th32ThreadID, + start_address=None, + state=None, + priority=int(entry.tpBasePri), + raw=entry.th32ThreadID, + ) + # THREADENTRY32 is reused across iterations — reset dwSize each + # time per Microsoft's sample code. + entry.dwSize = ctypes.sizeof(entry) + if not kernel32.Thread32Next(snapshot, ctypes.byref(entry)): + break + finally: + kernel32.CloseHandle(snapshot) + + def WriteProcessMemory( process_handle: int, address: int, diff --git a/PyMemoryEditor/win32/process.py b/PyMemoryEditor/win32/process.py index afb680d..f2df290 100644 --- a/PyMemoryEditor/win32/process.py +++ b/PyMemoryEditor/win32/process.py @@ -8,13 +8,16 @@ from ..enums import ScanTypesEnum from ..process import AbstractProcess from ..process.errors import ClosedProcess +from ..process.thread_info import ThreadInfo from .enums import ProcessOperationsEnum from .functions import ( CloseProcessHandle, GetMemoryRegions, GetProcessHandle, + GetThreads, ReadProcessMemory, + SearchAddressesByPattern, SearchAddressesByValue, SearchValuesByAddresses, WriteProcessMemory, @@ -78,6 +81,7 @@ def __init__( pid: Optional[int] = None, permission: Union[ProcessOperationsEnum, int] = DEFAULT_PERMISSION, case_sensitive: bool = False, + exact_match: bool = True, ): """ :param process_name: name of the target process. @@ -90,11 +94,14 @@ def __init__( a read-only handle, or pass PROCESS_ALL_ACCESS for full control. :param case_sensitive: when False (default on Windows), process_name matching ignores case to align with the OS convention. + :param exact_match: when False, ``process_name`` is matched as a + substring (e.g. ``"chrome"`` finds ``"chrome.exe"``). """ super().__init__( process_name=process_name, pid=pid, case_sensitive=case_sensitive, + exact_match=exact_match, ) self.__closed = False @@ -150,6 +157,11 @@ def get_memory_regions(self) -> Generator[dict, None, None]: self.__require_open() return GetMemoryRegions(self.__process_handle) + def get_threads(self) -> Generator[ThreadInfo, None, None]: + self.__require_open() + # Toolhelp32 takes a PID, not a handle — no PROCESS_* right needed. + return GetThreads(self.pid) + def search_by_addresses( self, pytype: Type[T], @@ -200,6 +212,24 @@ def search_by_value( memory_regions=memory_regions, ) + def search_by_pattern( + self, + pattern, + *, + byte_length: int = 0, + progress_information: bool = False, + memory_regions: Optional[Sequence[Dict]] = None, + ) -> Generator[Union[int, Tuple[int, dict]], None, None]: + self.__require_open() + self.__require_read() + return SearchAddressesByPattern( + self.__process_handle, + pattern, + byte_length=byte_length, + progress_information=progress_information, + memory_regions=memory_regions, + ) + def search_by_value_between( self, pytype: Type[T], diff --git a/PyMemoryEditor/win32/types.py b/PyMemoryEditor/win32/types.py index 38886f9..d7c3f94 100644 --- a/PyMemoryEditor/win32/types.py +++ b/PyMemoryEditor/win32/types.py @@ -61,3 +61,21 @@ class SYSTEM_INFO(Structure): if sizeof(c_void_p) == 8 else MEMORY_BASIC_INFORMATION_32 ) + + +# TH32CS_SNAPTHREAD flag for CreateToolhelp32Snapshot — used by get_threads(). +TH32CS_SNAPTHREAD = 0x00000004 + + +class THREADENTRY32(Structure): + """Layout matching the Win32 ``THREADENTRY32`` returned by Thread32First/Next.""" + + _fields_ = [ + ("dwSize", wintypes.DWORD), + ("cntUsage", wintypes.DWORD), + ("th32ThreadID", wintypes.DWORD), + ("th32OwnerProcessID", wintypes.DWORD), + ("tpBasePri", wintypes.LONG), + ("tpDeltaPri", wintypes.LONG), + ("dwFlags", wintypes.DWORD), + ] diff --git a/README.md b/README.md index 67ecf87..2449bec 100644 --- a/README.md +++ b/README.md @@ -20,14 +20,24 @@ reading, writing and searching values in the process memory. One unified API. Three operating systems. No C compiler. No native build step.

+

+ Tweak a value in a running game · inspect a live program's state · + harvest data straight from RAM — on Windows, Linux and macOS. +

+

Quick Start · Usage Guide · + Troubleshooting · Platform Notes · The App · Contributing

+

+ PyMemoryEditor app attached to a running process +

+

Runs on 🪟 Windows · 🐧 Linux · 🍎 macOS — 32-bit and 64-bit, with the same code on all three.

@@ -41,6 +51,8 @@ reading, writing and searching values in the process memory. | **Read & write memory** | Change live values on the fly — just like Cheat Engine, but in a few lines of Python. | | **Pure-Python via `ctypes`** | No compilation, no native wheels — `pip install` and you're done. | | **Scan modes** | Exact, not-exact, bigger / smaller (±equal), in-range, out-of-range. | +| **Pattern scan** | Byte signatures or regex — `grep` for process memory. | +| **Pointer chains** | Walk multi-level pointers (`[[base+0x10]+0x20]+0x30`) in one call. | | **Snapshot caching** | The Cheat-Engine "scan → refine → refine" loop, accelerated. | | **Bundled GUI app** | A full memory scanner ships in the box — just type `pymemoryeditor`. | @@ -72,9 +84,7 @@ plain Python types. Everything fits in a handful of lines: from PyMemoryEditor import OpenProcess with OpenProcess(process_name="example.exe") as process: - # Read a 4-byte int at a known address. - value = process.read_process_memory(0x0005000C, int) - print("Current value:", value) + value = 120 # Scan the whole process for every address holding that value. for address in process.search_by_value(int, 4, value): @@ -88,6 +98,38 @@ OpenProcess(process_name="notepad.exe") # by process name OpenProcess(pid=1234) # by PID ``` +### How it works in practice + +You rarely know the address of a value up front — you **find it by scanning**. +The typical loop is the same one Cheat Engine made famous: + +1. **Scan** for a value you can see (e.g. your health is `100`) — you get back + many candidate addresses. +2. **Let the value change** in the target (you take damage → `95`). +3. **Refine**: keep only the addresses that now hold the new value. Repeat until + one address remains — that's your value. +4. **Read, write or freeze** it. + +```python +with OpenProcess(process_name="game.exe") as process: + # 1. First scan — every address currently holding 100. + candidates = list(process.search_by_value(int, 4, 100)) + + # 3. After the value drops to 95 in-game, keep only the matches that agree. + survivors = [ + address + for address, value in process.search_by_addresses(int, 4, candidates) + if value == 95 + ] + + # 4. Overwrite the survivors back to a high value. + for address in survivors: + process.write_process_memory(address, int, 4, 9999) +``` + +> For big targets, cache the region map once and reuse it across scans — see +> [the refine-scan workflow](#the-refine-scan-workflow-recommended). + --- ## 📚 Usage Guide @@ -116,7 +158,7 @@ with OpenProcess(process_name="notepad.exe") as process: name = process.read_process_memory(address, str, 32) ``` -### Searching for a value +### 🔍 Searching for a value Look up a value anywhere in memory and stream every match: @@ -204,16 +246,73 @@ for address, value in process.search_by_addresses(int, 4, addresses_list): print("Address", hex(address), "holds the value", value) ``` -### Walking the memory map +### 🗺️ Exploring the memory regions -`get_memory_regions()` streams the address, size and metadata of every region the -target owns: +A process's address space is split into regions — contiguous blocks of memory, +each with its own size and permissions. `get_memory_regions()` streams the +address, size and metadata of every region the target owns: ```python for region in process.get_memory_regions(): print(hex(region["address"]), region["size"], region["struct"]) ``` +### 🧵 Listing the process threads + +`get_threads()` yields a `ThreadInfo` for every thread running inside the +target — useful for introspection (how many workers does it spawn? is the +main thread still alive?). `main_thread` is a shortcut to the lowest-id one. + +```python +for thread in process.get_threads(): + print(thread.tid, thread.state, thread.priority) + +print("Main thread:", process.main_thread.tid) +``` + +> `tid` is the OS-native thread id (POSIX TID on Linux, thread id on Windows, +> Mach port on macOS); `state` and `priority` are filled in where the platform +> exposes them and are `None` otherwise. + +### 🎯 Pattern scan — grep for process memory + +Pass a raw `bytes` regular expression and the scanner applies it directly to +memory, letting you locate *data* by its shape. The example below extracts +every email address held in the target's memory. + +```python +email = rb"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}" + +# byte_length is the maximum length of a single match (used to span chunk reads). +for address in process.search_by_pattern(email, byte_length=128): + raw = process.read_process_memory(address, bytes, 128) + print(address, raw.split(b"\x00", 1)[0].decode("ascii", "replace")) +``` + +Patterns also work the other way — for *code*. Absolute addresses shift between +builds, but **byte signatures remain stable**: provide an IDA-style hex string +with `?` wildcards and `search_by_pattern` returns every match, the same +technique Cheat Engine, IDA and Ghidra use to locate code that has moved. + +```python +for address in process.search_by_pattern("48 8B ? ? 00 00 89 ?"): + print(f"Match at 0x{address:X}") +``` + +### 🔗 Pointer chains — resolve addresses that change every run + +A multi-level pointer is a static base plus a series of offsets — +`module + offset → [+x] → [+y] → …`. Walk the whole chain in one line, then +read or write the final address as usual: + +```python +# module + 0x10F4F4 -> [+0x0] -> [+0x158] (a value behind a two-level pointer) +hp_address = process.resolve_pointer_chain(base + 0x10F4F4, [0x0, 0x158]) +hp = process.read_process_memory(hp_address, int, 4) +``` + +`ptr_size=4` for 32-bit targets, `ptr_size=8` (default) for 64-bit. + --- ## Platform Notes @@ -279,23 +378,18 @@ $ pymemoryeditor The app is a living demo of the library — it exercises every public surface (every `ScanTypesEnum` mode, every value type, scanning, refining, freezing values, the hex viewer, the memory map). If you're learning the API, it's the fastest way to see what's possible. -

- PyMemoryEditor app attached to a running process -

-
**✨ What you get out of the box** -- **Process picker** — list all running processes and pick by row, PID or name -- **Live scanner** — eight scan modes, value-between ranges, typed inputs +- **Scanner** — eight scan modes, ranges, byte-signature or regex search - **Refine workflow** — *First Scan → Next Scan → Next Scan…* like Cheat Engine -- **Value freezing** — pin a value so the target can't change it back -- **Memory map** — every region of the target, with R/W/X flags -- **Hex viewer** — auto-refreshing dump, write bytes back -- **Import/export** cheat tables as JSON +- **Cheat table** — freeze / write values, import/export as JSON +- **Pointer chains** — resolve multi-level pointers +- **Memory map** — regions with R/W/X flags +- **Hex viewer** — live dump with write-back @@ -330,6 +424,47 @@ Then launch the app by running `pymemoryeditor` from any terminal. The library i --- +## 🛟 Troubleshooting + +`PermissionError` when opening another process — the OS is denying access. +This is the most common first hurdle: +- **Windows:** run your terminal **as Administrator** for protected targets. +- **Linux:** run as root, or relax `ptrace_scope` + (`sudo sysctl kernel.yama.ptrace_scope=0`). Opening your own process always works. +- **macOS:** the Python binary must be signed with the + `com.apple.security.cs.debugger` entitlement (or SIP off + root). Opening the + *current* process always works — great for trying things out. + +`ProcessNotFoundError` — the name didn't match. Names are case-sensitive by +default; try `OpenProcess(process_name="chrome", exact_match=False, case_sensitive=False)` +for a fuzzy match, or pass the `pid=` directly. + +`AmbiguousProcessNameError` — more than one process matches. Pick one from the +listed PIDs and pass `pid=` instead. + +A scan returns nothing on Windows — region enumeration needs +`PROCESS_QUERY_INFORMATION`, which the default permission already includes. If you +passed a custom `permission=` mask, make sure that flag is in it. + +Reading an address gives garbage or raises `OSError` — the page may have been +freed between scan and read (normal during a live scan), or the value type / size +is wrong. Wrap one-off reads in `try/except OSError` and double-check the byte width. + +Need more detail? The library logs to a standard `logging` logger named +`"PyMemoryEditor"` (silent by default). Turn it on to see exactly which pages +the library skips during a scan and why: + +```python +import logging + +logging.basicConfig(level=logging.DEBUG) +logging.getLogger("PyMemoryEditor").setLevel(logging.DEBUG) +``` + +> The bundled app exposes the same stream in its **Log Console** (Tools → Log Console). + +--- + ## 🤝 Contributing Pull requests, bug reports and feature ideas are very welcome. Read diff --git a/assets/screenshots/app.png b/assets/screenshots/app.png index fba5627..ffe9d90 100644 Binary files a/assets/screenshots/app.png and b/assets/screenshots/app.png differ diff --git a/pyproject.toml b/pyproject.toml index 92f3444..4d5ccb1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -100,6 +100,11 @@ packages = ["PyMemoryEditor"] "PyMemoryEditor/py.typed" = "PyMemoryEditor/py.typed" "PyMemoryEditor/app/assets/icon.svg" = "PyMemoryEditor/app/assets/icon.svg" +# Maintainer-only tooling lives under scripts/ — keep it out of the source +# distribution (the wheel already only ships the PyMemoryEditor package). +[tool.hatch.build.targets.sdist] +exclude = ["scripts"] + [build-system] requires = ["hatchling"] build-backend = "hatchling.build" diff --git a/scripts/generate_app_screenshot.py b/scripts/generate_app_screenshot.py new file mode 100644 index 0000000..9a7ccad --- /dev/null +++ b/scripts/generate_app_screenshot.py @@ -0,0 +1,180 @@ +#!/usr/bin/env python3 +""" +Regenerate the README screenshot of the PyMemoryEditor app. + +This is a maintainer-only helper — it is intentionally kept out of the +published package (see the sdist/wheel excludes in ``pyproject.toml``). + +It launches the Qt app attached to *this* Python process (so it works on +every platform without special entitlements), stages a believable scan + +cheat-table scenario, and grabs the window to +``assets/screenshots/app.png``. + +Usage: + pip install "PyMemoryEditor[app]" + python scripts/generate_app_screenshot.py +""" +import os +import sys +from pathlib import Path + +from PySide6.QtCore import QTimer +from PySide6.QtWidgets import QApplication + +from PyMemoryEditor import OpenProcess +from PyMemoryEditor.app._icon import app_icon +from PyMemoryEditor.app.application import DEFAULT_THEME_ID, apply_theme +from PyMemoryEditor.app.cheat_entry import CheatEntry +from PyMemoryEditor.app.main_window import MainWindow +from PyMemoryEditor.app.value_types import find_spec + + +REPO_ROOT = Path(__file__).resolve().parent.parent +OUTPUT_PATH = str(REPO_ROOT / "assets" / "screenshots" / "app.png") + +rows = [ + (0x000055EF6A1C0008, 100, 87), + (0x000055EF6A1C0014, 100, 102), + (0x000055EF6A24007C, 100, 64), + (0x000055EF6A2400A0, 100, 113), + (0x000055EF6A300120, 100, 95), + (0x000055EF6A300C44, 100, 158), + (0x00007F8E1A4B8010, 100, 41), + (0x00007F8E1A4B8024, 100, 76), + (0x00007F8E1A500088, 100, 132), + (0x00007F8E1A5000F0, 100, 88), + (0x00007F8E1A5C0114, 100, 200), + (0x00007F8E1A6A8200, 100, 17), + (0x00007F8E1A6A82C0, 100, 99), + (0x00007F8E1A7B40A0, 100, 124), + (0x00007F8E1A7B41B4, 100, 53), + (0x00007F8E1A8C8050, 100, 181), + (0x00007F8E1A9D0140, 100, 72), + (0x00007F8E1AAE0090, 100, 145), + (0x00007F8E1ABF0030, 100, 28), +] * 19 + +def populate_results(window): + """Fill the Found Addresses table with believable refine-scan rows. + + Every current value is the scan target (100); the previous column varies + so the screenshot reads as "lots of candidates converged onto 100". + """ + spec = find_spec("4 Bytes (Int32)") + model = window._results_model + model.set_value_spec(spec) + + model.append_chunk([(addr, cur) for addr, cur, _ in rows]) + for i, (_, cur, prev) in enumerate(rows): + model._previous[i] = prev + model._values[i] = cur + model.layoutChanged.emit() + + window._results_label.setText(f"Found {len(rows)} addresses (showing all).") + window._scanner.set_has_results(True) + + +def populate_cheat_table(window): + """Add a few saved entries — one frozen — with believable last values.""" + cheat = window._cheat + entries = [ + ( + CheatEntry( + description="Player HP", + address=0x000055EF6A1C0008, + spec_label="4 Bytes (Int32)", + length=4, + frozen=True, + frozen_value=999, + ), + 999, + ), + ( + CheatEntry( + description="Ammo", + address=0x000055EF6A1C0014, + spec_label="4 Bytes (Int32)", + length=4, + ), + 42, + ), + ( + CheatEntry( + description="Coins", + address=0x000055EF6A24007C, + spec_label="4 Bytes (Int32)", + length=4, + ), + 1337, + ), + ( + CheatEntry( + description="Player Name", + address=0x00007F8E1A500088, + spec_label="String (UTF-8)", + length=16, + ), + "JeanExtreme002", + ), + ] + for entry, _ in entries: + cheat.add_entry(entry) + # Stamp last_value and refresh the value cells directly. Suspend + # cellChanged so setText() doesn't trigger a write into the fake addresses. + cheat._suspend_signals = True + try: + for row, (entry, value) in enumerate(entries): + entry.last_value = value + cheat._update_value_cell(row, entry) + finally: + cheat._suspend_signals = False + + +def main(): + app = QApplication.instance() or QApplication(sys.argv) + app.setApplicationName("PyMemoryEditor") + app.setApplicationDisplayName("PyMemoryEditor App") + app.setOrganizationName("PyMemoryEditor") + app.setWindowIcon(app_icon()) + apply_theme(app, DEFAULT_THEME_ID) + + process = OpenProcess(pid=os.getpid()) + window = MainWindow(process) + window.resize(1280, 780) + window.show() + + def shoot(): + app.processEvents() + window._scanner._value_edit.setText("100") + populate_results(window) + populate_cheat_table(window) + + # Show a completed scan in the progress bar and status bar. + window._progress.setValue(100) + window._status.showMessage(f"Checked 81,750/82,350, kept {len(rows)}") + + # Stop background timers so they don't overwrite the staged values + # between processEvents() and grab(). + try: + window._heartbeat.stop() + except Exception: + pass + try: + window._cheat._publish_timer.stop() + window._cheat._poller.stop() + except Exception: + pass + + app.processEvents() + pixmap = window.grab() + ok = pixmap.save(OUTPUT_PATH, "PNG") + print(f"saved={ok} path={OUTPUT_PATH} size={pixmap.width()}x{pixmap.height()}") + QTimer.singleShot(50, app.quit) + + # Let the event loop tick once so the window paints before we grab it. + QTimer.singleShot(400, shoot) + app.exec() + + +if __name__ == "__main__": + main() diff --git a/tests/test_logger.py b/tests/test_logger.py new file mode 100644 index 0000000..2642177 --- /dev/null +++ b/tests/test_logger.py @@ -0,0 +1,99 @@ +# -*- coding: utf-8 -*- + +""" +Tests for the package-level ``PyMemoryEditor`` logger. + +The logger is silent by default (NullHandler attached at import time) — these +tests attach a memory handler to capture emissions, then trigger code paths +that should log DEBUG events. +""" + +import logging + +import pytest + + +@pytest.fixture +def log_capture(): + """Attach a list-based handler to the PyMemoryEditor logger for the test.""" + records = [] + + class _ListHandler(logging.Handler): + def emit(self, record): + records.append(record) + + logger = logging.getLogger("PyMemoryEditor") + handler = _ListHandler(level=logging.DEBUG) + previous_level = logger.level + logger.addHandler(handler) + logger.setLevel(logging.DEBUG) + try: + yield records + finally: + logger.removeHandler(handler) + logger.setLevel(previous_level) + + +def test_logger_is_silent_by_default(): + """ + A fresh import attaches a NullHandler — calls to ``logger.debug(...)`` + must not surface anywhere unless the consumer adds a handler. + """ + logger = logging.getLogger("PyMemoryEditor") + # At least one handler — the NullHandler installed at import time. + assert logger.handlers, "expected at least one handler (NullHandler) attached" + # NullHandler swallows messages: emitting at DEBUG must not raise. + logger.debug("smoke test message") + + +def test_logger_module_exported(): + """The logger is also exported at the package top level.""" + import PyMemoryEditor + + assert PyMemoryEditor.logger is logging.getLogger("PyMemoryEditor") + + +def test_logger_captures_emit(log_capture): + """Sanity check: when a handler is attached, DEBUG records reach it.""" + logger = logging.getLogger("PyMemoryEditor") + logger.debug("hello %s", "world") + assert len(log_capture) == 1 + record = log_capture[0] + assert record.levelno == logging.DEBUG + assert record.getMessage() == "hello world" + + +def test_logger_emits_during_scanning_skip(log_capture): + """ + Triggering a real scan against the current process — which always has + a few unreadable regions — must log DEBUG entries from the iter_* + helpers when they skip those chunks. + """ + import os + import sys + + if sys.platform not in ("win32", "darwin") and not sys.platform.startswith("linux"): + pytest.skip("Platform not supported by PyMemoryEditor") + + from PyMemoryEditor import OpenProcess + + with OpenProcess(pid=os.getpid()) as p: + # A scan over the whole address space will touch unreadable chunks on + # every supported platform; the logger should record at least one + # skip. We don't iterate fully — just enough to ensure the loop runs. + for _ in p.search_by_value(int, 4, 0xCAFEBABE): + break + + # Some platforms may not log a skip on every run (a self-process scan may + # be entirely successful in a particular environment); when that happens, + # the test is inconclusive rather than failing. The point of this test is + # to confirm the logger is wired in — exercise it where possible without + # being flaky. + if not log_capture: + pytest.skip( + "no scan skips observed in this run — wiring is sound but this " + "environment didn't trip a transient skip" + ) + + debug_messages = [r.getMessage() for r in log_capture if r.levelno == logging.DEBUG] + assert debug_messages, "expected at least one DEBUG message during scanning" diff --git a/tests/test_partial_name_match.py b/tests/test_partial_name_match.py new file mode 100644 index 0000000..b5899d3 --- /dev/null +++ b/tests/test_partial_name_match.py @@ -0,0 +1,111 @@ +# -*- coding: utf-8 -*- + +""" +Tests for the ``exact_match`` flag on process-name lookup. Uses ``psutil`` +directly to derive the current process's real name from the OS, then +verifies that: + +* an exact-name lookup finds it, +* a partial (substring) lookup finds it, +* a substring lookup that cannot match anything returns nothing. + +Avoids relying on a specific executable being installed. +""" + +import os +import sys + +import psutil +import pytest + +from PyMemoryEditor.process.util import ( + get_process_id_by_process_name, + get_process_ids_by_process_name, +) + + +_OWN_PROCESS_NAME = psutil.Process(os.getpid()).name() or "" + + +@pytest.fixture(scope="module") +def own_name(): + """The OS-reported name of the test process (e.g. ``python3.12``).""" + if not _OWN_PROCESS_NAME: + pytest.skip("psutil cannot read this process's name on this platform") + return _OWN_PROCESS_NAME + + +def test_exact_match_finds_self(own_name): + pids = get_process_ids_by_process_name(own_name, exact_match=True) + assert os.getpid() in pids + + +def test_exact_match_does_not_find_substring(own_name): + """Substring of the name must NOT match in exact mode.""" + if len(own_name) <= 2: + pytest.skip("process name too short to test substring rejection") + substring = own_name[: len(own_name) // 2] + if substring == own_name: + pytest.skip("substring equals full name") + pids = get_process_ids_by_process_name(substring, exact_match=True) + assert os.getpid() not in pids + + +def test_partial_match_finds_self_by_substring(own_name): + """A leading substring of the name must match when exact_match=False.""" + if len(own_name) <= 2: + pytest.skip("process name too short to test substring matching") + substring = own_name[: max(2, len(own_name) // 2)] + pids = get_process_ids_by_process_name(substring, exact_match=False) + assert os.getpid() in pids + + +def test_partial_match_case_insensitive(own_name): + """Combined with case_sensitive=False, swapping case still matches.""" + swapped = own_name.swapcase() + if swapped == own_name: + pytest.skip("process name has no alphabetic characters") + pids = get_process_ids_by_process_name( + swapped, exact_match=False, case_sensitive=False + ) + assert os.getpid() in pids + + +def test_partial_match_no_results_for_garbage(): + """An impossible substring returns an empty list, not a false positive.""" + garbage = "definitely_not_a_real_process_name_zzzzzzz_42" + pids = get_process_ids_by_process_name(garbage, exact_match=False) + assert pids == [] + + +def test_get_single_returns_none_for_garbage(): + """The single-result helper returns None when no process matches.""" + garbage = "definitely_not_a_real_process_name_zzzzzzz_42" + assert ( + get_process_id_by_process_name(garbage, exact_match=False) is None + ) + + +@pytest.mark.skipif( + sys.platform not in ("win32", "darwin") and not sys.platform.startswith("linux"), + reason="Platform not supported by PyMemoryEditor", +) +def test_openprocess_accepts_exact_match_kwarg(own_name): + """OpenProcess plumbs exact_match through to the lookup.""" + from PyMemoryEditor import OpenProcess + + # Don't lean on a known unique name — only that the kwarg is accepted + # without raising TypeError and the resulting PID is ours when the name + # is unique enough to match exactly one process. + pids = get_process_ids_by_process_name(own_name, exact_match=True) + if len(pids) != 1: + pytest.skip( + "more than one process shares this name; OpenProcess would raise " + "AmbiguousProcessNameError — not what this test is checking" + ) + + process = OpenProcess(process_name=own_name, exact_match=True) + try: + assert process.pid == os.getpid() + finally: + process.close() diff --git a/tests/test_pattern_compile.py b/tests/test_pattern_compile.py new file mode 100644 index 0000000..b36e306 --- /dev/null +++ b/tests/test_pattern_compile.py @@ -0,0 +1,129 @@ +# -*- coding: utf-8 -*- + +""" +Unit tests for PyMemoryEditor.util.pattern.compile_pattern. + +These tests don't touch process memory — they validate the pattern compiler +in isolation so a regression in pattern parsing fails fast and clearly +instead of as a no-match in the scan loop. +""" + +import re + +import pytest + +from PyMemoryEditor.util.pattern import compile_pattern + + +def test_ida_style_basic(): + """A plain hex string compiles and matches its own bytes.""" + pattern, length = compile_pattern("48 8B 90") + assert length == 3 + assert pattern.search(b"\x48\x8B\x90") is not None + assert pattern.search(b"\x48\x8B\x91") is None + + +def test_ida_style_wildcard_single_question(): + """`?` is a one-byte wildcard.""" + pattern, length = compile_pattern("48 ? 90") + assert length == 3 + assert pattern.search(b"\x48\x00\x90") is not None + assert pattern.search(b"\x48\xFF\x90") is not None + # Wildcard must NOT match more or fewer than one byte. + assert pattern.search(b"\x48\x90") is None + + +def test_ida_style_wildcard_double_question(): + """`??` is the alternate one-byte wildcard syntax — same semantics as `?`.""" + pattern, length = compile_pattern("48 ?? 90") + assert length == 3 + assert pattern.search(b"\x48\xAB\x90") is not None + + +def test_ida_style_multiple_wildcards(): + pattern, length = compile_pattern("DE AD ? ? EF") + assert length == 5 + assert pattern.search(b"\xDE\xAD\xBE\xEF\xEF") is not None + assert pattern.search(b"\xDE\xAD\x00\x00\xEF") is not None + + +def test_ida_style_irregular_whitespace(): + """Multiple spaces/tabs/newlines between tokens are fine.""" + pattern, length = compile_pattern("48 8B\t90\n00") + assert length == 4 + assert pattern.search(b"\x48\x8B\x90\x00") is not None + + +def test_ida_style_lowercase_hex(): + """Lower-case hex digits should be accepted as well as upper-case.""" + pattern, length = compile_pattern("de ad be ef") + assert length == 4 + assert pattern.search(b"\xDE\xAD\xBE\xEF") is not None + + +def test_ida_style_escapes_regex_specials(): + """ + Bytes like 0x5C (``\\``), 0x28 (``(``) or 0x2E (``.``) are regex + meta chars. The compiler must escape them so the pattern still matches + the literal byte and not interpret it as regex syntax. + """ + # 0x5C, 0x28 and 0x2E are all regex specials. + pattern, length = compile_pattern("5C 28 2E") + assert length == 3 + assert pattern.search(b"\x5C\x28\x2E") is not None + # If 0x2E was left as a regex `.`, the next match would falsely succeed: + assert pattern.search(b"\x5C\x28\x00") is None + + +def test_ida_style_rejects_bad_token(): + # Token has the right length but contains non-hex digits — the compiler + # routes this through bytes.fromhex which raises a different message + # than the "wrong length" branch above. Match either of the two ValueError + # phrasings to keep the test resilient to minor message changes. + with pytest.raises(ValueError, match="(not valid hex|not two hex digits)"): + compile_pattern("48 8B Z9") + + +def test_ida_style_rejects_single_digit_token(): + with pytest.raises(ValueError, match="not two hex digits"): + compile_pattern("48 8 90") + + +def test_ida_style_rejects_empty(): + with pytest.raises(ValueError, match="Empty pattern"): + compile_pattern(" ") + + +def test_bytes_regex_requires_explicit_length(): + """Bytes regex without ``byte_length=`` must error — the source length + is not a reliable proxy for the matched length.""" + with pytest.raises(ValueError, match="byte_length"): + compile_pattern(rb"\x48\x8B") + + +def test_bytes_regex_with_explicit_length(): + """Bytes regex compiles with DOTALL — ``.`` matches any byte (incl. 0x0A).""" + pattern, length = compile_pattern(rb"\x48\x8B..", byte_length=4) + assert length == 4 + assert pattern.search(b"\x48\x8B\x0A\x0B") is not None + # The dot must even match newline bytes (DOTALL is the key): + assert pattern.search(b"\x48\x8B\n\n") is not None + + +def test_compiled_pattern_passthrough(): + """Passing an already-compiled re.Pattern returns it unchanged.""" + precompiled = re.compile(rb"\xDE\xAD", re.DOTALL) + pattern, length = compile_pattern(precompiled, byte_length=2) + assert pattern is precompiled + assert length == 2 + + +def test_compiled_pattern_requires_explicit_length(): + precompiled = re.compile(rb"\xDE\xAD", re.DOTALL) + with pytest.raises(ValueError, match="byte_length"): + compile_pattern(precompiled) + + +def test_rejects_unsupported_input_type(): + with pytest.raises(TypeError, match="Pattern must be"): + compile_pattern(12345) # type: ignore[arg-type] diff --git a/tests/test_pattern_scan.py b/tests/test_pattern_scan.py new file mode 100644 index 0000000..4b426db --- /dev/null +++ b/tests/test_pattern_scan.py @@ -0,0 +1,118 @@ +# -*- coding: utf-8 -*- + +""" +Integration tests for ``process.search_by_pattern`` against the current +process. Plants a known marker on the test's own stack, then verifies that +the scanner finds it via: + +* a literal IDA-style pattern, +* an IDA-style pattern with wildcards, +* a raw bytes regex (``re.DOTALL``). + +All tests use ``OpenProcess(pid=os.getpid())`` — no special privilege needed +on any of the three supported platforms. +""" + +import ctypes +import os +import sys + +import pytest + +if sys.platform not in ("win32", "darwin") and not sys.platform.startswith("linux"): + pytest.skip("Platform not supported by PyMemoryEditor", allow_module_level=True) + + +from PyMemoryEditor import OpenProcess # noqa: E402 + + +# A reasonably distinctive marker — eight bytes give us enough entropy that +# random other matches in the test process's address space are unlikely. +_MARKER = b"\x90\x90\xDE\xAD\xBE\xEF\xCA\xFE" + + +@pytest.fixture +def planted_marker(): + """Create a buffer holding the marker; yield (address, marker_bytes).""" + buf = ctypes.create_string_buffer(_MARKER, len(_MARKER)) + yield ctypes.addressof(buf), _MARKER + # `buf` stays alive for the entire test fixture scope — GC collects it + # after the test returns, at which point the address may be reused. + + +def test_pattern_scan_finds_exact_marker(planted_marker): + address, _ = planted_marker + with OpenProcess(pid=os.getpid()) as process: + hits = list(process.search_by_pattern("90 90 DE AD BE EF CA FE")) + assert address in hits, ( + "expected scan to find the planted marker at 0x%X among %d hits" + % (address, len(hits)) + ) + + +def test_pattern_scan_with_wildcards(planted_marker): + """Wildcards on the middle bytes still locate the marker.""" + address, _ = planted_marker + with OpenProcess(pid=os.getpid()) as process: + hits = list(process.search_by_pattern("90 90 ? ? BE EF CA FE")) + assert address in hits + + +def test_pattern_scan_bytes_regex(planted_marker): + """Bytes regex with explicit byte_length.""" + address, _ = planted_marker + with OpenProcess(pid=os.getpid()) as process: + hits = list( + process.search_by_pattern( + rb"\xDE\xAD\xBE\xEF\xCA\xFE", byte_length=6 + ) + ) + # The 6-byte slice starts 2 bytes into the marker. + assert (address + 2) in hits + + +def test_pattern_scan_progress_information(planted_marker): + """``progress_information=True`` yields (address, info) tuples.""" + address, _ = planted_marker + with OpenProcess(pid=os.getpid()) as process: + items = list( + process.search_by_pattern( + "90 90 DE AD BE EF CA FE", progress_information=True + ) + ) + assert items, "expected at least one hit" + + # Validate shape of the first tuple. + addr, info = items[0] + assert isinstance(addr, int) + assert "progress" in info + assert "memory_total" in info + assert 0.0 <= info["progress"] <= 1.0 + + # The planted marker must be one of the addresses. + assert any(item[0] == address for item in items) + + +def test_pattern_scan_no_match(): + """A pattern that cannot exist in our address space yields no hits.""" + # 16 random-looking bytes; the chance of a coincidence is effectively zero. + with OpenProcess(pid=os.getpid()) as process: + hits = list( + process.search_by_pattern( + "DE AD BE EF C0 DE F0 0D FE ED BA BE 13 37 C0 DE" + ) + ) + assert hits == [] + + +def test_pattern_scan_accepts_memory_regions_snapshot(planted_marker): + """Passing a pre-built region snapshot must produce the same matches.""" + address, _ = planted_marker + with OpenProcess(pid=os.getpid()) as process: + snapshot = process.snapshot_memory_regions() + hits_with_snapshot = list( + process.search_by_pattern( + "90 90 DE AD BE EF CA FE", memory_regions=snapshot + ) + ) + assert address in hits_with_snapshot diff --git a/tests/test_pointer_chain.py b/tests/test_pointer_chain.py new file mode 100644 index 0000000..bdc12fe --- /dev/null +++ b/tests/test_pointer_chain.py @@ -0,0 +1,166 @@ +# -*- coding: utf-8 -*- + +""" +Tests for ``AbstractProcess.resolve_pointer_chain``. Builds a chain of +ctypes pointers on the test's own heap and verifies that walking the chain +recovers the final address — the same operation Cheat Engine performs to +locate values that survive a process restart. +""" + +import ctypes +import os +import sys + +import pytest + +if sys.platform not in ("win32", "darwin") and not sys.platform.startswith("linux"): + pytest.skip("Platform not supported by PyMemoryEditor", allow_module_level=True) + + +from PyMemoryEditor import OpenProcess # noqa: E402 + + +@pytest.fixture +def process(): + """Open `OpenProcess` against the current process for the whole module.""" + handle = OpenProcess(pid=os.getpid()) + try: + yield handle + finally: + handle.close() + + +def test_resolve_empty_offsets_dereferences_once(process): + """ + With ``offsets=[]``, resolve_pointer_chain must dereference base once + and return the resulting pointer — no further walking, no offset added. + """ + target = ctypes.c_int(0xCAFEF00D) + pointer_holder = ctypes.c_uint64(ctypes.addressof(target)) + + resolved = process.resolve_pointer_chain(ctypes.addressof(pointer_holder), []) + + assert resolved == ctypes.addressof(target) + + +def test_resolve_two_level_chain(process): + """``[base] → [ptr1+0] → ptr2+0`` resolves to the deepest address.""" + target = ctypes.c_int(0xDEADBEEF) + level1 = ctypes.c_uint64(ctypes.addressof(target)) + level0 = ctypes.c_uint64(ctypes.addressof(level1)) + + # Walk two levels, no extra offset on the last hop. + resolved = process.resolve_pointer_chain( + ctypes.addressof(level0), [0, 0] + ) + + assert resolved == ctypes.addressof(target) + # And reading at the resolved address yields the value we planted. + value = process.read_process_memory(resolved, int, 4) + assert (value & 0xFFFFFFFF) == 0xDEADBEEF + + +def test_resolve_chain_with_offsets(process): + """A chain whose last hop adds a non-zero offset returns base+offset.""" + # Layout: a struct-like buffer with two int fields, plus a pointer to it. + pair = (ctypes.c_int * 2)(0x11111111, 0x22222222) + pointer_holder = ctypes.c_uint64(ctypes.addressof(pair)) + + # offsets[-1] = 4 — points to the second int, *without* extra dereference. + resolved = process.resolve_pointer_chain( + ctypes.addressof(pointer_holder), [4] + ) + + second_int_address = ctypes.addressof(pair) + 4 + assert resolved == second_int_address + + value = process.read_process_memory(resolved, int, 4) + assert (value & 0xFFFFFFFF) == 0x22222222 + + +def test_resolve_four_level_chain_with_large_offsets(process): + """ + Walk a four-level pointer chain whose every offset is greater than 100 — + same shape as the deep cheat-table dumps people share for modern games + (e.g. ``"game.exe"+0x10F4F4 -> [+0x68] -> [+0x90] -> [+0xC8] -> [+0x158]``). + + Each level is a 256-byte buffer holding the pointer to the next level at + a non-trivial offset, so the test exercises pointer arithmetic that has + *nothing* in common with the "all-zeros" happy path. + """ + level_size = 256 + offsets = [104, 128, 152, 200] # all > 100, all distinct, none aligned with each other + target_value = 0xABCD1234 + + # Innermost level holds the value we want to recover, parked at the deep + # offset rather than at the start of the buffer. + level4 = (ctypes.c_uint8 * level_size)() + ctypes.memmove( + ctypes.addressof(level4) + offsets[3], + ctypes.byref(ctypes.c_int32(target_value)), + 4, + ) + + # Each intermediate level stores the *address* of the next level at the + # appropriate offset. memmove of a c_uint64 writes 8 raw bytes in native + # byte order — exactly what the resolver will read back. + level3 = (ctypes.c_uint8 * level_size)() + ctypes.memmove( + ctypes.addressof(level3) + offsets[2], + ctypes.byref(ctypes.c_uint64(ctypes.addressof(level4))), + 8, + ) + + level2 = (ctypes.c_uint8 * level_size)() + ctypes.memmove( + ctypes.addressof(level2) + offsets[1], + ctypes.byref(ctypes.c_uint64(ctypes.addressof(level3))), + 8, + ) + + level1 = (ctypes.c_uint8 * level_size)() + ctypes.memmove( + ctypes.addressof(level1) + offsets[0], + ctypes.byref(ctypes.c_uint64(ctypes.addressof(level2))), + 8, + ) + + # Base just holds a raw pointer to level1 (no offset on the first hop). + base = ctypes.c_uint64(ctypes.addressof(level1)) + + resolved = process.resolve_pointer_chain(ctypes.addressof(base), offsets) + + expected = ctypes.addressof(level4) + offsets[3] + assert resolved == expected, ( + "4-level chain resolved to 0x%X, expected 0x%X" % (resolved, expected) + ) + + # And the value at the final address must be the one we planted. + value = process.read_process_memory(resolved, int, 4) + assert (value & 0xFFFFFFFF) == target_value + + +def test_resolve_rejects_invalid_ptr_size(process): + target = ctypes.c_int(1) + pointer_holder = ctypes.c_uint64(ctypes.addressof(target)) + with pytest.raises(ValueError, match="ptr_size"): + process.resolve_pointer_chain( + ctypes.addressof(pointer_holder), [], ptr_size=5 + ) + + +def test_resolve_unsigned_pointer_decoding(process): + """ + A pointer whose top bit is set must be returned as a positive int, not + sign-extended. We don't always have a pointer in the upper half of the + address space available, but we can fabricate one by writing arbitrary + bytes and checking the decoding directly via ptr_size=4 to keep the + bit-pattern controllable on every platform. + """ + raw = ctypes.c_uint32(0xFFFFFFFF) + resolved = process.resolve_pointer_chain( + ctypes.addressof(raw), [], ptr_size=4 + ) + # If sign-extension leaked through, ``resolved`` would be negative. + assert resolved == 0xFFFFFFFF + assert resolved > 0 diff --git a/tests/test_thread_enumeration.py b/tests/test_thread_enumeration.py new file mode 100644 index 0000000..069c621 --- /dev/null +++ b/tests/test_thread_enumeration.py @@ -0,0 +1,134 @@ +# -*- coding: utf-8 -*- + +""" +Cross-platform tests for ``AbstractProcess.get_threads`` and the +``main_thread`` property. Spawns a few extra Python threads in the test +process so we can verify that the enumeration sees more than just one. +""" + +import os +import sys +import threading +import time + +import pytest + +if sys.platform not in ("win32", "darwin") and not sys.platform.startswith("linux"): + pytest.skip("Platform not supported by PyMemoryEditor", allow_module_level=True) + + +from PyMemoryEditor import OpenProcess, ThreadInfo # noqa: E402 + + +@pytest.fixture +def extra_threads(): + """ + Start a handful of long-running Python threads, yield to the test, then + signal them to exit. Each spinning thread bumps the kernel-visible + thread count for the duration of the test — without that we'd be + racing with the GIL/GC threads to assert ``len(threads) > 1`` and the + test could pass coincidentally. + """ + stop = threading.Event() + + def _spin(): + while not stop.is_set(): + time.sleep(0.05) + + threads = [threading.Thread(target=_spin, daemon=True) for _ in range(3)] + for t in threads: + t.start() + + # Give the OS a moment to actually schedule the new threads — without + # this delay, Toolhelp32 (Windows) and task_threads (macOS) sometimes + # snapshot before the threads have been registered. + time.sleep(0.1) + + try: + yield threads + finally: + stop.set() + for t in threads: + t.join(timeout=2.0) + + +def test_get_threads_returns_thread_infos(extra_threads): + """Each yielded item must be a ``ThreadInfo`` with a non-negative tid.""" + with OpenProcess(pid=os.getpid()) as process: + threads = list(process.get_threads()) + + assert threads, "expected at least one thread" + for info in threads: + assert isinstance(info, ThreadInfo) + assert isinstance(info.tid, int) + assert info.tid >= 0 + + +def test_get_threads_sees_extra_threads(extra_threads): + """The spinning threads from the fixture must show up in the enumeration.""" + with OpenProcess(pid=os.getpid()) as process: + threads = list(process.get_threads()) + + # The exact mapping from Python ``Thread.ident`` to the OS-visible tid is + # platform-specific (and on macOS, Mach port names don't match POSIX + # tids), so we can't compare ident-to-tid directly. Instead we assert + # the *count*: with 3 extra spinning threads, the main thread, plus + # Python's GC / runtime helpers, the count must comfortably exceed 1. + assert len(threads) > 1, ( + "expected get_threads() to see at least the main + 1 spinning thread; " + "got %d" % len(threads) + ) + + +def test_get_threads_yields_unique_tids(extra_threads): + """``tid`` should be unique within the snapshot.""" + with OpenProcess(pid=os.getpid()) as process: + tids = [t.tid for t in process.get_threads()] + + assert len(tids) == len(set(tids)), "duplicate tids in enumeration" + + +def test_main_thread_is_smallest_tid(extra_threads): + """``main_thread`` returns the ThreadInfo with the smallest tid.""" + with OpenProcess(pid=os.getpid()) as process: + all_threads = list(process.get_threads()) + main = process.main_thread + + assert main is not None + assert main.tid == min(t.tid for t in all_threads) + + +def test_thread_info_is_hashable_and_comparable(): + """``ThreadInfo`` is a frozen dataclass — usable as dict keys / set members.""" + info_a = ThreadInfo(tid=42) + info_b = ThreadInfo(tid=42) + info_c = ThreadInfo(tid=43) + + # Frozen dataclasses get __eq__ + __hash__ from the dataclass decorator. + assert info_a == info_b + assert info_a != info_c + # ``raw`` is excluded from equality (compare=False) so two entries from + # different snapshots with the same tid still compare equal. + info_d = ThreadInfo(tid=42, raw="snapshot-1") + info_e = ThreadInfo(tid=42, raw="snapshot-2") + assert info_d == info_e + + # And hashable: + {info_a, info_b, info_c} + + +@pytest.mark.skipif( + not sys.platform.startswith("linux"), + reason="Linux-specific: /proc//task//stat exposes state/priority", +) +def test_linux_threads_populate_state_and_priority(extra_threads): + """Linux backend should fill ``state`` and ``priority`` from /proc/.../stat.""" + with OpenProcess(pid=os.getpid()) as process: + threads = list(process.get_threads()) + + # Not every entry — the file can vanish between listdir and open — but + # at least one should have parsed values. + has_state = any(t.state is not None for t in threads) + has_priority = any(t.priority is not None for t in threads) + assert has_state, "expected at least one thread to have a state field" + assert has_priority, "expected at least one thread to have a priority field"