forked from jan/mem_edit
Compare commits
25 commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
46e9456fd4 | ||
| f3154e443d | |||
| ef1a39152c | |||
| c29be9f429 | |||
| 5a032da984 | |||
| 6ab295fc26 | |||
| 6913f73db4 | |||
| 9759645f92 | |||
| 0632b205ab | |||
|
|
5c75da31d5 | ||
|
|
bd6c22ca1d | ||
| 260d67bf81 | |||
| 4deaa41d7e | |||
| 8b5d5af95b | |||
| e8c6c4f74c | |||
| e842f81575 | |||
| 49a7c21ed2 | |||
| 6321d4221c | |||
| d0bbed8db1 | |||
| 83e105dc30 | |||
| 53b1b1ade8 | |||
| 571ecc7a95 | |||
| 5021d5fb9a | |||
| 522999cd61 | |||
| 3b766be616 |
10 changed files with 342 additions and 194 deletions
1
.gitignore
vendored
1
.gitignore
vendored
|
|
@ -4,4 +4,5 @@ __pycache__
|
||||||
*.pyc
|
*.pyc
|
||||||
|
|
||||||
*.egg-info/
|
*.egg-info/
|
||||||
|
build/
|
||||||
dist/
|
dist/
|
||||||
|
|
|
||||||
|
|
@ -1,2 +1,3 @@
|
||||||
include README.md
|
include README.md
|
||||||
include LICENSE.md
|
include LICENSE.md
|
||||||
|
include mem_edit/VERSION
|
||||||
|
|
|
||||||
|
|
@ -18,19 +18,19 @@
|
||||||
## Installation
|
## Installation
|
||||||
|
|
||||||
**Dependencies:**
|
**Dependencies:**
|
||||||
* python 3 (written and tested with 3.5)
|
* python 3 (written and tested with 3.7)
|
||||||
* ctypes
|
* ctypes
|
||||||
* typing (for type annotations)
|
* typing (for type annotations)
|
||||||
|
|
||||||
|
|
||||||
Install with pip, from PyPI (preferred):
|
Install with pip, from PyPI (preferred):
|
||||||
```bash
|
```bash
|
||||||
pip install mem_edit
|
pip3 install mem_edit
|
||||||
```
|
```
|
||||||
|
|
||||||
Install with pip from git repository
|
Install with pip from git repository
|
||||||
```bash
|
```bash
|
||||||
pip install git+https://mpxd.net/code/jan/mem_edit.git@release
|
pip3 install git+https://mpxd.net/code/jan/mem_edit.git@release
|
||||||
```
|
```
|
||||||
|
|
||||||
|
|
||||||
|
|
|
||||||
4
mem_edit/VERSION.py
Normal file
4
mem_edit/VERSION.py
Normal file
|
|
@ -0,0 +1,4 @@
|
||||||
|
""" VERSION defintion. THIS FILE IS MANUALLY PARSED BY setup.py and REQUIRES A SPECIFIC FORMAT """
|
||||||
|
__version__ = '''
|
||||||
|
0.6
|
||||||
|
'''.strip()
|
||||||
|
|
@ -18,6 +18,9 @@ from .utils import MemEditError
|
||||||
|
|
||||||
__author__ = 'Jan Petykiewicz'
|
__author__ = 'Jan Petykiewicz'
|
||||||
|
|
||||||
|
from .VERSION import __version__
|
||||||
|
version = __version__ # legacy compatibility
|
||||||
|
|
||||||
|
|
||||||
system = platform.system()
|
system = platform.system()
|
||||||
if system == 'Windows':
|
if system == 'Windows':
|
||||||
|
|
|
||||||
|
|
@ -2,17 +2,17 @@
|
||||||
Abstract class for cross-platform memory editing.
|
Abstract class for cross-platform memory editing.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from typing import List, Tuple
|
from typing import List, Tuple, Optional, Union, Generator
|
||||||
from abc import ABCMeta, abstractmethod
|
from abc import ABCMeta, abstractmethod
|
||||||
from contextlib import contextmanager
|
from contextlib import contextmanager
|
||||||
import copy
|
import copy
|
||||||
import ctypes
|
import ctypes
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
from .utils import ctypes_buffer_t, search_buffer, ctypes_equal
|
from . import utils
|
||||||
|
from .utils import ctypes_buffer_t
|
||||||
|
|
||||||
|
|
||||||
logging.basicConfig(level=logging.INFO)
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -22,8 +22,8 @@ class Process(metaclass=ABCMeta):
|
||||||
(i.e., by reading from or writing to the memory used by a given process).
|
(i.e., by reading from or writing to the memory used by a given process).
|
||||||
|
|
||||||
The static methods
|
The static methods
|
||||||
Process.list_available_pids()
|
`Process.list_available_pids()`
|
||||||
Process.get_pid_by_name(executable_filename)
|
`Process.get_pid_by_name(executable_filename)`
|
||||||
can be used to help find the process id (pid) of the target process. They are
|
can be used to help find the process id (pid) of the target process. They are
|
||||||
provided for convenience only; it is probably better to use the tools built
|
provided for convenience only; it is probably better to use the tools built
|
||||||
in to your operating system to discover the pid of the specific process you
|
in to your operating system to discover the pid of the specific process you
|
||||||
|
|
@ -31,18 +31,19 @@ class Process(metaclass=ABCMeta):
|
||||||
|
|
||||||
Once you have found the pid, you are ready to construct an instance of Process
|
Once you have found the pid, you are ready to construct an instance of Process
|
||||||
and use it to read and write to memory. Once you are done with the process,
|
and use it to read and write to memory. Once you are done with the process,
|
||||||
use .close() to free up the process for access by other debuggers etc.
|
use `.close()` to free up the process for access by other debuggers etc.
|
||||||
|
```
|
||||||
p = Process(1239)
|
p = Process(1239)
|
||||||
p.close()
|
p.close()
|
||||||
|
```
|
||||||
|
|
||||||
To read/write to memory, first create a buffer using ctypes:
|
To read/write to memory, first create a buffer using ctypes:
|
||||||
|
```
|
||||||
buffer0 = (ctypes.c_byte * 5)(39, 50, 03, 40, 30)
|
buffer0 = (ctypes.c_byte * 5)(39, 50, 03, 40, 30)
|
||||||
buffer1 = ctypes.c_ulong()
|
buffer1 = ctypes.c_ulong()
|
||||||
|
```
|
||||||
and then use
|
and then use
|
||||||
|
```
|
||||||
p.write_memory(0x2fe, buffer0)
|
p.write_memory(0x2fe, buffer0)
|
||||||
|
|
||||||
val0 = p.read_memory(0x220, buffer0)[:]
|
val0 = p.read_memory(0x220, buffer0)[:]
|
||||||
|
|
@ -50,52 +51,52 @@ class Process(metaclass=ABCMeta):
|
||||||
val1a = p.read_memory(0x149, buffer1).value
|
val1a = p.read_memory(0x149, buffer1).value
|
||||||
val2b = buffer1.value
|
val2b = buffer1.value
|
||||||
assert(val1a == val2b)
|
assert(val1a == val2b)
|
||||||
|
```
|
||||||
|
|
||||||
Searching for a value can be done in a number of ways:
|
Searching for a value can be done in a number of ways:
|
||||||
Search a list of addresses:
|
Search a list of addresses:
|
||||||
found_addresses = p.search_addresses([0x1020, 0x1030], buffer0)
|
`found_addresses = p.search_addresses([0x1020, 0x1030], buffer0)`
|
||||||
Search the entire memory space:
|
Search the entire memory space:
|
||||||
found_addresses = p.search_all_memory(buffer0, writeable_only=False)
|
`found_addresses = p.search_all_memory(buffer0, writeable_only=False)`
|
||||||
|
|
||||||
You can also get a list of which regions in memory are mapped (readable):
|
You can also get a list of which regions in memory are mapped (readable):
|
||||||
regions = p.list_mapped_regions(writeable_only=False)
|
`regions = p.list_mapped_regions(writeable_only=False)`
|
||||||
|
|
||||||
which can be used along with search_buffer(...) to re-create .search_all_memory(...):
|
which can be used along with search_buffer(...) to re-create .search_all_memory(...):
|
||||||
|
```
|
||||||
found = []
|
found = []
|
||||||
for region_start, region_stop in regions:
|
for region_start, region_stop in regions:
|
||||||
region_buffer = (ctypes.c_byte * (region_stop - region_start))()
|
region_buffer = (ctypes.c_byte * (region_stop - region_start))()
|
||||||
p.read_memory(region_start, region_buffer)
|
p.read_memory(region_start, region_buffer)
|
||||||
found += utils.search_buffer(ctypes.c_ulong(123456790), region_buffer)
|
found += utils.search_buffer(ctypes.c_ulong(123456790), region_buffer)
|
||||||
|
```
|
||||||
Other useful methods include the context manager, implemented as a static method:
|
Other useful methods include the context manager, implemented as a static method:
|
||||||
|
```
|
||||||
with Process.open_process(pid) as p:
|
with Process.open_process(pid) as p:
|
||||||
# use p here, no need to call p.close()
|
# use p here, no need to call p.close()
|
||||||
|
```
|
||||||
.get_path(), which reports the path of the executable file which was used
|
.get_path(), which reports the path of the executable file which was used
|
||||||
to start the process:
|
to start the process:
|
||||||
|
```
|
||||||
executable_path = p.get_path()
|
executable_path = p.get_path()
|
||||||
|
```
|
||||||
and deref_struct_pointer, which takes a pointer to a struct and reads out the struct members:
|
and deref_struct_pointer, which takes a pointer to a struct and reads out the struct members:
|
||||||
|
```
|
||||||
# struct is a list of (offset, buffer) pairs
|
# struct is a list of (offset, buffer) pairs
|
||||||
struct_defintion = [(0x0, ctypes.c_ulong()),
|
struct_defintion = [(0x0, ctypes.c_ulong()),
|
||||||
(0x20, ctypes.c_byte())]
|
(0x20, ctypes.c_byte())]
|
||||||
values = p.deref_struct_pointer(0x0feab4, struct_defintion)
|
values = p.deref_struct_pointer(0x0feab4, struct_defintion)
|
||||||
|
```
|
||||||
which is shorthand for
|
which is shorthand for
|
||||||
|
```
|
||||||
struct_addr = p.read_memory(0x0feab4, ctypes.c_void_p())
|
struct_addr = p.read_memory(0x0feab4, ctypes.c_void_p())
|
||||||
values = [p.read_memory(struct_addr + 0x0, ctypes.c_ulong()),
|
values = [p.read_memory(struct_addr + 0x0, ctypes.c_ulong()),
|
||||||
p.read_memory(struct_addr + 0x20, ctypes.c_byte())]
|
p.read_memory(struct_addr + 0x20, ctypes.c_byte())]
|
||||||
|
```
|
||||||
=================
|
=================
|
||||||
|
|
||||||
Putting all this together, a simple program which alters a magic number in the only running
|
Putting all this together, a simple program which alters a magic number in the only running
|
||||||
instance of 'magic.exe' might look like this:
|
instance of 'magic.exe' might look like this:
|
||||||
|
```
|
||||||
import ctypes
|
import ctypes
|
||||||
from mem_edit import Process
|
from mem_edit import Process
|
||||||
|
|
||||||
|
|
@ -106,9 +107,9 @@ class Process(metaclass=ABCMeta):
|
||||||
addrs = p.search_all_memory(magic_number)
|
addrs = p.search_all_memory(magic_number)
|
||||||
assert(len(addrs) == 1)
|
assert(len(addrs) == 1)
|
||||||
p.write_memory(addrs[0], ctypes.c_ulong(42))
|
p.write_memory(addrs[0], ctypes.c_ulong(42))
|
||||||
|
```
|
||||||
Searching for a value which changes:
|
Searching for a value which changes:
|
||||||
|
```
|
||||||
pid = Process.get_pid_by_name('monitor_me.exe')
|
pid = Process.get_pid_by_name('monitor_me.exe')
|
||||||
with Process.open_process(pid) as p:
|
with Process.open_process(pid) as p:
|
||||||
addrs = p.search_all_memory(ctypes.c_int(40))
|
addrs = p.search_all_memory(ctypes.c_int(40))
|
||||||
|
|
@ -117,18 +118,19 @@ class Process(metaclass=ABCMeta):
|
||||||
print('Found addresses:')
|
print('Found addresses:')
|
||||||
for addr in filtered_addrs:
|
for addr in filtered_addrs:
|
||||||
print(hex(addr))
|
print(hex(addr))
|
||||||
|
```
|
||||||
"""
|
"""
|
||||||
|
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def __init__(self, process_id: int):
|
def __init__(self, process_id: int):
|
||||||
"""
|
"""
|
||||||
Constructing a Process object prepares the process with specified process_id for
|
Constructing a Process object prepares the process with specified process_id for
|
||||||
memory editing. Finding the process_id for the process you want to edit is often
|
memory editing. Finding the `process_id` for the process you want to edit is often
|
||||||
easiest using os-specific tools (or by launching the process yourself, e.g. with
|
easiest using os-specific tools (or by launching the process yourself, e.g. with
|
||||||
subprocess.Popen(...)).
|
`subprocess.Popen(...)`).
|
||||||
|
|
||||||
:param process_id: Process id (pid) of the target process
|
Args:
|
||||||
|
process_id: Process id (pid) of the target process
|
||||||
"""
|
"""
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
@ -139,7 +141,7 @@ class Process(metaclass=ABCMeta):
|
||||||
letting other debuggers attach to it instead.
|
letting other debuggers attach to it instead.
|
||||||
|
|
||||||
This function should be called after you are done working with the process
|
This function should be called after you are done working with the process
|
||||||
and will no longer need it. See the Process.open_process(...) context
|
and will no longer need it. See the `Process.open_process(...)` context
|
||||||
manager to avoid having to call this function yourself.
|
manager to avoid having to call this function yourself.
|
||||||
"""
|
"""
|
||||||
pass
|
pass
|
||||||
|
|
@ -147,38 +149,45 @@ class Process(metaclass=ABCMeta):
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def write_memory(self, base_address: int, write_buffer: ctypes_buffer_t):
|
def write_memory(self, base_address: int, write_buffer: ctypes_buffer_t):
|
||||||
"""
|
"""
|
||||||
Write the given buffer to the process's address space, starting at base_address.
|
Write the given buffer to the process's address space, starting at `base_address`.
|
||||||
|
|
||||||
:param base_address: The address to write at, in the process's address space.
|
Args:
|
||||||
:param write_buffer: A ctypes object, for example, ctypes.c_ulong(48),
|
base_address: The address to write at, in the process's address space.
|
||||||
(ctypes.c_byte * 3)(43, 21, 0xff), or a subclass of ctypes.Structure,
|
write_buffer: A ctypes object, for example, `ctypes.c_ulong(48)`,
|
||||||
which will be written into memory starting at base_address.
|
`(ctypes.c_byte * 3)(43, 21, 0xff)`, or a subclass of `ctypes.Structure`,
|
||||||
|
which will be written into memory starting at `base_address`.
|
||||||
"""
|
"""
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def read_memory(self, base_address: int, read_buffer: ctypes_buffer_t) -> ctypes_buffer_t:
|
def read_memory(self, base_address: int, read_buffer: ctypes_buffer_t) -> ctypes_buffer_t:
|
||||||
"""
|
"""
|
||||||
Read into the given buffer from the process's address space, starting at base_address.
|
Read into the given buffer from the process's address space, starting at `base_address`.
|
||||||
|
|
||||||
:param base_address: The address to read from, in the process's address space.
|
Args:
|
||||||
:param read_buffer: A ctypes object, for example. ctypes.c_ulong(),
|
base_address: The address to read from, in the process's address space.
|
||||||
(ctypes.c_byte * 3)(), or a subclass of ctypes.Structure, which will be
|
read_buffer: A `ctypes` object, for example. `ctypes.c_ulong()`,
|
||||||
overwritten with the contents of the process's memory starting at base_address.
|
`(ctypes.c_byte * 3)()`, or a subclass of `ctypes.Structure`, which will be
|
||||||
:returns: read_buffer is returned as well as being overwritten.
|
overwritten with the contents of the process's memory starting at `base_address`.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
`read_buffer` is returned as well as being overwritten.
|
||||||
"""
|
"""
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def list_mapped_regions(self, writeable_only=True) -> List[Tuple[int, int]]:
|
def list_mapped_regions(self, writeable_only=True) -> List[Tuple[int, int]]:
|
||||||
"""
|
"""
|
||||||
Return a list of (start_address, stop_address) for the regions of the address space
|
Return a list of `(start_address, stop_address)` for the regions of the address space
|
||||||
accessible to (readable and possibly writable by) the process.
|
accessible to (readable and possibly writable by) the process.
|
||||||
By default, this function does not return non-writeable regions.
|
By default, this function does not return non-writeable regions.
|
||||||
|
|
||||||
:param writeable_only: If True, only return regions which are also writeable.
|
Args:
|
||||||
Default true.
|
writeable_only: If `True`, only return regions which are also writeable.
|
||||||
:return: List of (start_address, stop_address) for each accessible memory region.
|
Default `True`.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
List of `(start_address, stop_address)` for each accessible memory region.
|
||||||
"""
|
"""
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
@ -187,7 +196,8 @@ class Process(metaclass=ABCMeta):
|
||||||
"""
|
"""
|
||||||
Return the path to the executable file which was run to start this process.
|
Return the path to the executable file which was run to start this process.
|
||||||
|
|
||||||
:return: A string containing the path.
|
Returns:
|
||||||
|
A string containing the path.
|
||||||
"""
|
"""
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
@ -197,13 +207,14 @@ class Process(metaclass=ABCMeta):
|
||||||
"""
|
"""
|
||||||
Return a list of all process ids (pids) accessible on this system.
|
Return a list of all process ids (pids) accessible on this system.
|
||||||
|
|
||||||
:return: List of running process ids.
|
Returns:
|
||||||
|
List of running process ids.
|
||||||
"""
|
"""
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def get_pid_by_name(target_name: str) -> int or None:
|
def get_pid_by_name(target_name: str) -> Optional[int]:
|
||||||
"""
|
"""
|
||||||
Attempt to return the process id (pid) of a process which was run with an executable
|
Attempt to return the process id (pid) of a process which was run with an executable
|
||||||
file with the provided name. If no process is found, return None.
|
file with the provided name. If no process is found, return None.
|
||||||
|
|
@ -214,7 +225,12 @@ class Process(metaclass=ABCMeta):
|
||||||
Don't rely on this method if you can possibly avoid it, since it makes no
|
Don't rely on this method if you can possibly avoid it, since it makes no
|
||||||
attempt to confirm that it found a unique process and breaks trivially (e.g. if the
|
attempt to confirm that it found a unique process and breaks trivially (e.g. if the
|
||||||
executable file is renamed).
|
executable file is renamed).
|
||||||
:return: Process id (pid) of a process with the provided name, or None.
|
|
||||||
|
Args:
|
||||||
|
target_name: Name of the process to find the PID for
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Process id (pid) of a process with the provided name, or `None`.
|
||||||
"""
|
"""
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
@ -224,73 +240,115 @@ class Process(metaclass=ABCMeta):
|
||||||
) -> List[ctypes_buffer_t]:
|
) -> List[ctypes_buffer_t]:
|
||||||
"""
|
"""
|
||||||
Take a pointer to a struct and read out the struct members:
|
Take a pointer to a struct and read out the struct members:
|
||||||
|
```
|
||||||
struct_defintion = [(0x0, ctypes.c_ulong()),
|
struct_defintion = [(0x0, ctypes.c_ulong()),
|
||||||
(0x20, ctypes.c_byte())]
|
(0x20, ctypes.c_byte())]
|
||||||
values = p.deref_struct_pointer(0x0feab4, struct_defintion)
|
values = p.deref_struct_pointer(0x0feab4, struct_defintion)
|
||||||
|
```
|
||||||
which is shorthand for
|
which is shorthand for
|
||||||
|
```
|
||||||
struct_addr = p.read_memory(0x0feab4, ctypes.c_void_p())
|
struct_addr = p.read_memory(0x0feab4, ctypes.c_void_p())
|
||||||
values = [p.read_memory(struct_addr + 0x0, ctypes.c_ulong()),
|
values = [p.read_memory(struct_addr + 0x0, ctypes.c_ulong()),
|
||||||
p.read_memory(struct_addr + 0x20, ctypes.c_byte())]
|
p.read_memory(struct_addr + 0x20, ctypes.c_byte())]
|
||||||
|
```
|
||||||
|
|
||||||
:param base_address: Address at which the struct pointer is located.
|
Args:
|
||||||
:param targets: List of (offset, read_buffer) pairs which will be read from the struct.
|
base_address: Address at which the struct pointer is located.
|
||||||
:return: List of read values corresponding to the provided targets.
|
targets: List of `(offset, read_buffer)` pairs which will be read from the struct.
|
||||||
|
|
||||||
|
Return:
|
||||||
|
List of read values corresponding to the provided targets.
|
||||||
"""
|
"""
|
||||||
base = self.read_memory(base_address, ctypes.c_void_p()).value
|
base = self.read_memory(base_address, ctypes.c_void_p()).value
|
||||||
values = [self.read_memory(base + offset, buffer) for offset, buffer in targets]
|
values = [self.read_memory(base + offset, buffer) for offset, buffer in targets]
|
||||||
return values
|
return values
|
||||||
|
|
||||||
def search_addresses(self, addresses: List[int], needle_buffer: ctypes_buffer_t) -> List[int]:
|
def search_addresses(self,
|
||||||
|
addresses: List[int],
|
||||||
|
needle_buffer: ctypes_buffer_t,
|
||||||
|
verbatim: bool = True,
|
||||||
|
) -> List[int]:
|
||||||
"""
|
"""
|
||||||
Search for the provided value at each of the provided addresses, and return the addresses
|
Search for the provided value at each of the provided addresses, and return the addresses
|
||||||
where it is found.
|
where it is found.
|
||||||
|
|
||||||
:param addresses: List of addresses which should be probed.
|
Args:
|
||||||
:param needle_buffer: The value to search for. This should be a ctypes object of the same
|
addresses: List of addresses which should be probed.
|
||||||
sorts as used by .read_memory(...), which will be compared to the contents of
|
needle_buffer: The value to search for. This should be a `ctypes` object of the same
|
||||||
|
sorts as used by `.read_memory(...)`, which will be compared to the contents of
|
||||||
memory at each of the given addresses.
|
memory at each of the given addresses.
|
||||||
:return: List of addresses where the needle_buffer was found.
|
verbatim: If `True`, perform bitwise comparison when searching for `needle_buffer`.
|
||||||
|
If `False`, perform `utils.ctypes_equal`-based comparison. Default `True`.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
List of addresses where the `needle_buffer` was found.
|
||||||
"""
|
"""
|
||||||
found = []
|
found = []
|
||||||
read_buffer = copy.copy(needle_buffer)
|
read_buffer = copy.copy(needle_buffer)
|
||||||
|
|
||||||
|
if verbatim:
|
||||||
|
def compare(a, b):
|
||||||
|
return bytes(read_buffer) == bytes(needle_buffer)
|
||||||
|
else:
|
||||||
|
compare = utils.ctypes_equal
|
||||||
|
|
||||||
for address in addresses:
|
for address in addresses:
|
||||||
read = self.read_memory(address, read_buffer)
|
self.read_memory(address, read_buffer)
|
||||||
if ctypes_equal(needle_buffer, read):
|
if compare(needle_buffer, read_buffer):
|
||||||
found.append(address)
|
found.append(address)
|
||||||
return found
|
return found
|
||||||
|
|
||||||
def search_all_memory(self, needle_buffer, writeable_only=True) -> List[int]:
|
def search_all_memory(self,
|
||||||
|
needle_buffer: ctypes_buffer_t,
|
||||||
|
writeable_only: bool = True,
|
||||||
|
verbatim: bool = True,
|
||||||
|
) -> List[int]:
|
||||||
"""
|
"""
|
||||||
Search the entire memory space accessible to the process for the provided value.
|
Search the entire memory space accessible to the process for the provided value.
|
||||||
|
|
||||||
:param needle_buffer: The value to search for. This should be a ctypes object of the same
|
Args:
|
||||||
sorts as used by .read_memory(...), which will be compared to the contents of
|
needle_buffer: The value to search for. This should be a ctypes object of the same
|
||||||
|
sorts as used by `.read_memory(...)`, which will be compared to the contents of
|
||||||
memory at each accessible address.
|
memory at each accessible address.
|
||||||
:param writeable_only: If True, only search regions where the process has write access.
|
writeable_only: If `True`, only search regions where the process has write access.
|
||||||
:return: List of addresses where the needle_buffer was found.
|
Default `True`.
|
||||||
|
verbatim: If `True`, perform bitwise comparison when searching for `needle_buffer`.
|
||||||
|
If `False`, perform `utils.ctypes_equal-based` comparison. Default `True`.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
List of addresses where the `needle_buffer` was found.
|
||||||
"""
|
"""
|
||||||
found = []
|
found = []
|
||||||
|
if verbatim:
|
||||||
|
search = utils.search_buffer_verbatim
|
||||||
|
else:
|
||||||
|
search = utils.search_buffer
|
||||||
|
|
||||||
for start, stop in self.list_mapped_regions(writeable_only):
|
for start, stop in self.list_mapped_regions(writeable_only):
|
||||||
try:
|
try:
|
||||||
region_buffer = (ctypes.c_byte * (stop - start))()
|
region_buffer = (ctypes.c_byte * (stop - start))()
|
||||||
self.read_memory(start, region_buffer)
|
self.read_memory(start, region_buffer)
|
||||||
found += [offset + start for offset in search_buffer(needle_buffer, region_buffer)]
|
found += [offset + start for offset in search(needle_buffer, region_buffer)]
|
||||||
except OSError:
|
except OSError:
|
||||||
logger.error('Failed to read in range 0x{} - 0x{}'.format(start, stop))
|
logger.error('Failed to read in range 0x{} - 0x{}'.format(start, stop))
|
||||||
return found
|
return found
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
@contextmanager
|
@contextmanager
|
||||||
def open_process(cls, process_id: int) -> 'Process':
|
def open_process(cls, process_id: int) -> Generator['Process', None, None]:
|
||||||
"""
|
"""
|
||||||
Context manager which automatically closes the constructed Process:
|
Context manager which automatically closes the constructed Process:
|
||||||
|
```
|
||||||
with Process.open_process(2394) as p:
|
with Process.open_process(2394) as p:
|
||||||
# use p here
|
# use p here
|
||||||
# no need to run p.close()
|
# no need to run p.close()
|
||||||
|
```
|
||||||
|
|
||||||
:param process_id: Process id (pid), passed to the Process constructor.
|
Args:
|
||||||
:return: Constructed Process object.
|
process_id: Process id (pid), passed to the Process constructor.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Constructed Process object.
|
||||||
"""
|
"""
|
||||||
process = cls(process_id)
|
process = cls(process_id)
|
||||||
yield process
|
yield process
|
||||||
|
|
|
||||||
|
|
@ -2,7 +2,7 @@
|
||||||
Implementation of Process class for Linux
|
Implementation of Process class for Linux
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from typing import List, Tuple
|
from typing import List, Tuple, Optional
|
||||||
from os import strerror
|
from os import strerror
|
||||||
import os
|
import os
|
||||||
import os.path
|
import os.path
|
||||||
|
|
@ -15,7 +15,6 @@ from .abstract import Process as AbstractProcess
|
||||||
from .utils import ctypes_buffer_t, MemEditError
|
from .utils import ctypes_buffer_t, MemEditError
|
||||||
|
|
||||||
|
|
||||||
logging.basicConfig(level=logging.INFO)
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -59,7 +58,9 @@ class Process(AbstractProcess):
|
||||||
|
|
||||||
def close(self):
|
def close(self):
|
||||||
os.kill(self.pid, signal.SIGSTOP)
|
os.kill(self.pid, signal.SIGSTOP)
|
||||||
|
os.waitpid(self.pid, 0)
|
||||||
ptrace(ptrace_commands['PTRACE_DETACH'], self.pid, 0, 0)
|
ptrace(ptrace_commands['PTRACE_DETACH'], self.pid, 0, 0)
|
||||||
|
os.kill(self.pid, signal.SIGCONT)
|
||||||
self.pid = None
|
self.pid = None
|
||||||
|
|
||||||
def write_memory(self, base_address: int, write_buffer: ctypes_buffer_t):
|
def write_memory(self, base_address: int, write_buffer: ctypes_buffer_t):
|
||||||
|
|
@ -91,17 +92,17 @@ class Process(AbstractProcess):
|
||||||
return pids
|
return pids
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def get_pid_by_name(target_name: str) -> int or None:
|
def get_pid_by_name(target_name: str) -> Optional[int]:
|
||||||
for pid in Process.list_available_pids():
|
for pid in Process.list_available_pids():
|
||||||
try:
|
try:
|
||||||
logger.info('Checking name for pid {}'.format(pid))
|
logger.debug('Checking name for pid {}'.format(pid))
|
||||||
with open('/proc/{}/cmdline'.format(pid), 'rb') as cmdline:
|
with open('/proc/{}/cmdline'.format(pid), 'rb') as cmdline:
|
||||||
path = cmdline.read().decode().split('\x00')[0]
|
path = cmdline.read().decode().split('\x00')[0]
|
||||||
except FileNotFoundError:
|
except FileNotFoundError:
|
||||||
continue
|
continue
|
||||||
|
|
||||||
name = os.path.basename(path)
|
name = os.path.basename(path)
|
||||||
logger.info('Name was "{}"'.format(name))
|
logger.debug('Name was "{}"'.format(name))
|
||||||
if path is not None and name == target_name:
|
if path is not None and name == target_name:
|
||||||
return pid
|
return pid
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -12,24 +12,57 @@ Utility functions and types:
|
||||||
ctypes_equal(a, b)
|
ctypes_equal(a, b)
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from typing import List
|
from typing import List, Union
|
||||||
import ctypes
|
import ctypes
|
||||||
|
|
||||||
|
|
||||||
ctypes_buffer_t = ctypes._SimpleCData or ctypes.Array or ctypes.Structure or ctypes.Union
|
ctypes_buffer_t = Union[ctypes._SimpleCData, ctypes.Array, ctypes.Structure, ctypes.Union]
|
||||||
|
|
||||||
|
|
||||||
class MemEditError(Exception):
|
class MemEditError(Exception):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
def search_buffer(needle_buffer: ctypes_buffer_t, haystack_buffer: ctypes_buffer_t) -> List[int]:
|
def search_buffer_verbatim(needle_buffer: ctypes_buffer_t,
|
||||||
|
haystack_buffer: ctypes_buffer_t,
|
||||||
|
) -> List[int]:
|
||||||
"""
|
"""
|
||||||
Search for a buffer inside another buffer.
|
Search for a buffer inside another buffer, using a direct (bitwise) comparison
|
||||||
|
|
||||||
:param needle_buffer: Buffer to search for.
|
Args:
|
||||||
:param haystack_buffer: Buffer to search in.
|
needle_buffer: Buffer to search for.
|
||||||
:return: List of offsets where the needle_buffer was found.
|
haystack_buffer: Buffer to search in.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
List of offsets where the `needle_buffer` was found.
|
||||||
|
"""
|
||||||
|
found = []
|
||||||
|
|
||||||
|
haystack = bytes(haystack_buffer)
|
||||||
|
needle = bytes(needle_buffer)
|
||||||
|
|
||||||
|
start = 0
|
||||||
|
result = haystack.find(needle, start)
|
||||||
|
while start < len(haystack) and result != -1:
|
||||||
|
found.append(result)
|
||||||
|
start = result + 1
|
||||||
|
result = haystack.find(needle, start)
|
||||||
|
return found
|
||||||
|
|
||||||
|
|
||||||
|
def search_buffer(needle_buffer: ctypes_buffer_t,
|
||||||
|
haystack_buffer: ctypes_buffer_t,
|
||||||
|
) -> List[int]:
|
||||||
|
"""
|
||||||
|
Search for a buffer inside another buffer, using `ctypes_equal` for comparison.
|
||||||
|
Much slower than `search_buffer_verbatim`.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
needle_buffer: Buffer to search for.
|
||||||
|
haystack_buffer: Buffer to search in.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
List of offsets where the needle_buffer was found.
|
||||||
"""
|
"""
|
||||||
found = []
|
found = []
|
||||||
read_type = type(needle_buffer)
|
read_type = type(needle_buffer)
|
||||||
|
|
@ -40,7 +73,9 @@ def search_buffer(needle_buffer: ctypes_buffer_t, haystack_buffer: ctypes_buffer
|
||||||
return found
|
return found
|
||||||
|
|
||||||
|
|
||||||
def ctypes_equal(a: ctypes_buffer_t, b: ctypes_buffer_t) -> bool:
|
def ctypes_equal(a: ctypes_buffer_t,
|
||||||
|
b: ctypes_buffer_t,
|
||||||
|
) -> bool:
|
||||||
"""
|
"""
|
||||||
Check if the values stored inside two ctypes buffers are equal.
|
Check if the values stored inside two ctypes buffers are equal.
|
||||||
"""
|
"""
|
||||||
|
|
|
||||||
|
|
@ -2,7 +2,7 @@
|
||||||
Implementation of Process class for Windows
|
Implementation of Process class for Windows
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from typing import List, Tuple
|
from typing import List, Tuple, Optional
|
||||||
from math import floor
|
from math import floor
|
||||||
from os import strerror
|
from os import strerror
|
||||||
import os.path
|
import os.path
|
||||||
|
|
@ -14,7 +14,6 @@ from .abstract import Process as AbstractProcess
|
||||||
from .utils import ctypes_buffer_t, MemEditError
|
from .utils import ctypes_buffer_t, MemEditError
|
||||||
|
|
||||||
|
|
||||||
logging.basicConfig(level=logging.INFO)
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -67,19 +66,54 @@ mem_types = {
|
||||||
'MEM_PRIVATE': 0x20000,
|
'MEM_PRIVATE': 0x20000,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
# C struct for VirtualQueryEx
|
# C struct for VirtualQueryEx
|
||||||
class MEMORY_BASIC_INFORMATION(ctypes.Structure):
|
class MEMORY_BASIC_INFORMATION32(ctypes.Structure):
|
||||||
_fields_ = [
|
_fields_ = [
|
||||||
('BaseAddress', ctypes.c_void_p),
|
('BaseAddress', ctypes.wintypes.DWORD),
|
||||||
('AllocationBase', ctypes.c_void_p),
|
('AllocationBase', ctypes.wintypes.DWORD),
|
||||||
('AllocationProtect', ctypes.wintypes.DWORD),
|
('AllocationProtect', ctypes.wintypes.DWORD),
|
||||||
('RegionSize', ctypes.wintypes.UINT),
|
('RegionSize', ctypes.wintypes.DWORD),
|
||||||
('State', ctypes.wintypes.DWORD),
|
('State', ctypes.wintypes.DWORD),
|
||||||
('Protect', ctypes.wintypes.DWORD),
|
('Protect', ctypes.wintypes.DWORD),
|
||||||
('Type', ctypes.wintypes.DWORD),
|
('Type', ctypes.wintypes.DWORD),
|
||||||
]
|
]
|
||||||
|
|
||||||
|
class MEMORY_BASIC_INFORMATION64(ctypes.Structure):
|
||||||
|
_fields_ = [
|
||||||
|
('BaseAddress', ctypes.c_ulonglong),
|
||||||
|
('AllocationBase', ctypes.c_ulonglong),
|
||||||
|
('AllocationProtect', ctypes.wintypes.DWORD),
|
||||||
|
('__alignment1', ctypes.wintypes.DWORD),
|
||||||
|
('RegionSize', ctypes.c_ulonglong),
|
||||||
|
('State', ctypes.wintypes.DWORD),
|
||||||
|
('Protect', ctypes.wintypes.DWORD),
|
||||||
|
('Type', ctypes.wintypes.DWORD),
|
||||||
|
('__alignment2', ctypes.wintypes.DWORD),
|
||||||
|
]
|
||||||
|
|
||||||
|
PTR_SIZE = ctypes.sizeof(ctypes.c_void_p)
|
||||||
|
if PTR_SIZE == 8: # 64-bit python
|
||||||
|
MEMORY_BASIC_INFORMATION = MEMORY_BASIC_INFORMATION64
|
||||||
|
elif PTR_SIZE == 4: # 32-bit python
|
||||||
|
MEMORY_BASIC_INFORMATION = MEMORY_BASIC_INFORMATION32
|
||||||
|
|
||||||
|
ctypes.windll.kernel32.VirtualQueryEx.argtypes = [
|
||||||
|
ctypes.wintypes.HANDLE,
|
||||||
|
ctypes.wintypes.LPCVOID,
|
||||||
|
ctypes.c_void_p,
|
||||||
|
ctypes.c_size_t]
|
||||||
|
ctypes.windll.kernel32.ReadProcessMemory.argtypes = [
|
||||||
|
ctypes.wintypes.HANDLE,
|
||||||
|
ctypes.wintypes.LPCVOID,
|
||||||
|
ctypes.c_void_p,
|
||||||
|
ctypes.c_size_t,
|
||||||
|
ctypes.c_void_p]
|
||||||
|
ctypes.windll.kernel32.WriteProcessMemory.argtypes = [
|
||||||
|
ctypes.wintypes.HANDLE,
|
||||||
|
ctypes.wintypes.LPCVOID,
|
||||||
|
ctypes.c_void_p,
|
||||||
|
ctypes.c_size_t,
|
||||||
|
ctypes.c_void_p]
|
||||||
|
|
||||||
# C struct for GetSystemInfo
|
# C struct for GetSystemInfo
|
||||||
class SYSTEM_INFO(ctypes.Structure):
|
class SYSTEM_INFO(ctypes.Structure):
|
||||||
|
|
@ -89,7 +123,7 @@ class SYSTEM_INFO(ctypes.Structure):
|
||||||
('dwPageSize', ctypes.wintypes.DWORD),
|
('dwPageSize', ctypes.wintypes.DWORD),
|
||||||
('lpMinimumApplicationAddress', ctypes.c_void_p),
|
('lpMinimumApplicationAddress', ctypes.c_void_p),
|
||||||
('lpMaximumApplicationAddress', ctypes.c_void_p),
|
('lpMaximumApplicationAddress', ctypes.c_void_p),
|
||||||
('dwActiveProcessorMask', ctypes.wintypes.DWORD),
|
('dwActiveProcessorMask', ctypes.c_void_p),
|
||||||
('dwNumberOfProcessors', ctypes.wintypes.DWORD),
|
('dwNumberOfProcessors', ctypes.wintypes.DWORD),
|
||||||
('dwProcessorType', ctypes.wintypes.DWORD),
|
('dwProcessorType', ctypes.wintypes.DWORD),
|
||||||
('dwAllocationGranularity', ctypes.wintypes.DWORD),
|
('dwAllocationGranularity', ctypes.wintypes.DWORD),
|
||||||
|
|
@ -154,8 +188,7 @@ class Process(AbstractProcess):
|
||||||
rval = ctypes.windll.psapi.GetProcessImageFileNameA(
|
rval = ctypes.windll.psapi.GetProcessImageFileNameA(
|
||||||
self.process_handle,
|
self.process_handle,
|
||||||
name_buffer,
|
name_buffer,
|
||||||
max_path_len
|
max_path_len)
|
||||||
)
|
|
||||||
|
|
||||||
if rval > 0:
|
if rval > 0:
|
||||||
return name_buffer.value.decode()
|
return name_buffer.value.decode()
|
||||||
|
|
@ -192,19 +225,21 @@ class Process(AbstractProcess):
|
||||||
return pids[:num_returned]
|
return pids[:num_returned]
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def get_pid_by_name(target_name: str) -> int or None:
|
def get_pid_by_name(target_name: str) -> Optional[int]:
|
||||||
for pid in Process.list_available_pids():
|
for pid in Process.list_available_pids():
|
||||||
try:
|
try:
|
||||||
logger.info('Checking name for pid {}'.format(pid))
|
logger.debug('Checking name for pid {}'.format(pid))
|
||||||
with Process.open_process(pid) as process:
|
with Process.open_process(pid) as process:
|
||||||
path = process.get_path()
|
path = process.get_path()
|
||||||
|
|
||||||
name = os.path.basename(path)
|
name = os.path.basename(path)
|
||||||
logger.info('Name was "{}"'.format(name))
|
logger.debug('Name was "{}"'.format(name))
|
||||||
if path is not None and name == target_name:
|
if path is not None and name == target_name:
|
||||||
return pid
|
return pid
|
||||||
except ValueError:
|
except ValueError:
|
||||||
pass
|
pass
|
||||||
|
except MemEditError as err:
|
||||||
|
logger.debug(repr(err))
|
||||||
|
|
||||||
logger.info('Found no process with name {}'.format(target_name))
|
logger.info('Found no process with name {}'.format(target_name))
|
||||||
return None
|
return None
|
||||||
|
|
@ -229,8 +264,7 @@ class Process(AbstractProcess):
|
||||||
self.process_handle,
|
self.process_handle,
|
||||||
address,
|
address,
|
||||||
mbi_ptr,
|
mbi_ptr,
|
||||||
mbi_size,
|
mbi_size)
|
||||||
)
|
|
||||||
|
|
||||||
if success != mbi_size:
|
if success != mbi_size:
|
||||||
if success == 0:
|
if success == 0:
|
||||||
|
|
@ -245,10 +279,11 @@ class Process(AbstractProcess):
|
||||||
page_ptr = start
|
page_ptr = start
|
||||||
while page_ptr < stop:
|
while page_ptr < stop:
|
||||||
page_info = get_mem_info(page_ptr)
|
page_info = get_mem_info(page_ptr)
|
||||||
if page_info.Type == mem_types['MEM_PRIVATE'] and \
|
if (page_info.Type == mem_types['MEM_PRIVATE']
|
||||||
page_info.State == mem_states['MEM_COMMIT'] and \
|
and page_info.State == mem_states['MEM_COMMIT']
|
||||||
page_info.Protect & page_protections['PAGE_READABLE'] != 0 and \
|
and page_info.Protect & page_protections['PAGE_READABLE'] != 0
|
||||||
(page_info.Protect & page_protections['PAGE_READWRITEABLE'] != 0 or not writeable_only):
|
and (page_info.Protect & page_protections['PAGE_READWRITEABLE'] != 0
|
||||||
|
or not writeable_only)):
|
||||||
regions.append((page_ptr, page_ptr + page_info.RegionSize))
|
regions.append((page_ptr, page_ptr + page_info.RegionSize))
|
||||||
page_ptr += page_info.RegionSize
|
page_ptr += page_info.RegionSize
|
||||||
|
|
||||||
|
|
|
||||||
20
setup.py
20
setup.py
|
|
@ -1,12 +1,21 @@
|
||||||
#!/usr/bin/env python
|
#!/usr/bin/env python3
|
||||||
|
|
||||||
from setuptools import setup, find_packages
|
from setuptools import setup, find_packages
|
||||||
|
|
||||||
|
|
||||||
|
with open('README.md', 'rt') as f:
|
||||||
|
long_description = f.read()
|
||||||
|
|
||||||
|
with open('mem_edit/VERSION.py', 'rt') as f:
|
||||||
|
version = f.readlines()[2].strip()
|
||||||
|
|
||||||
setup(name='mem_edit',
|
setup(name='mem_edit',
|
||||||
version='0.1',
|
version=version,
|
||||||
description='Multi-platform library for memory editing',
|
description='Multi-platform library for memory editing',
|
||||||
|
long_description=long_description,
|
||||||
|
long_description_content_type='text/markdown',
|
||||||
author='Jan Petykiewicz',
|
author='Jan Petykiewicz',
|
||||||
author_email='anewusername@gmail.com',
|
author_email='jan@mpxd.net',
|
||||||
url='https://mpxd.net/code/jan/mem_edit',
|
url='https://mpxd.net/code/jan/mem_edit',
|
||||||
keywords=[
|
keywords=[
|
||||||
'memory',
|
'memory',
|
||||||
|
|
@ -26,7 +35,6 @@ setup(name='mem_edit',
|
||||||
'trainer',
|
'trainer',
|
||||||
],
|
],
|
||||||
classifiers=[
|
classifiers=[
|
||||||
'Programming Language :: Python',
|
|
||||||
'Programming Language :: Python :: 3',
|
'Programming Language :: Python :: 3',
|
||||||
'Development Status :: 4 - Beta',
|
'Development Status :: 4 - Beta',
|
||||||
'Environment :: Other Environment',
|
'Environment :: Other Environment',
|
||||||
|
|
@ -42,8 +50,10 @@ setup(name='mem_edit',
|
||||||
'Topic :: Utilities',
|
'Topic :: Utilities',
|
||||||
],
|
],
|
||||||
packages=find_packages(),
|
packages=find_packages(),
|
||||||
|
package_data={
|
||||||
|
'mem_edit': []
|
||||||
|
},
|
||||||
install_requires=[
|
install_requires=[
|
||||||
'ctypes',
|
|
||||||
'typing',
|
'typing',
|
||||||
],
|
],
|
||||||
extras_require={
|
extras_require={
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue