Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions docs/api/process.rst
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,24 @@ Module
:undoc-members:
:show-inheritance:

Linux (platform-specific)
-------------------------

.. autoclass:: memobj.process.linux.process.LinuxProcess
:members:
:undoc-members:
:show-inheritance:

.. autoclass:: memobj.process.linux.process.LinuxMemoryRegion
:members:
:undoc-members:
:show-inheritance:

.. autoclass:: memobj.process.linux.module.LinuxModule
:members:
:undoc-members:
:show-inheritance:

Windows (platform-specific)
---------------------------

Expand Down
3 changes: 2 additions & 1 deletion memobj/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

from .allocation import Allocation, Allocator
from .object import MemoryObject
from .process import Process, WindowsProcess
from .process import Process, LinuxProcess, WindowsProcess
from . import property

logger = logging.getLogger(__name__)
Expand All @@ -17,6 +17,7 @@
"Allocator",
"MemoryObject",
"Process",
"LinuxProcess",
"WindowsProcess",
"property",
]
13 changes: 6 additions & 7 deletions memobj/hook.py
Original file line number Diff line number Diff line change
Expand Up @@ -282,7 +282,7 @@ def hook(self):
bitness = 64 if self.process.process_64_bit else 32

allocation_size = sum(map(len, hook_instructions))
hook_allocation = self.allocate_variable("hook_site", allocation_size)
hook_allocation = self.allocate_variable("hook_site", allocation_size, preferred_start=target_address)
hook_code = instructions_to_code(
hook_instructions, hook_allocation.address, bitness=bitness
)
Expand Down Expand Up @@ -351,7 +351,7 @@ def get_hook_tail(self, jump_address: int) -> tuple[list[Instruction], int]:
else:
bitness = 32

decoder = Decoder(bitness, search_bytes, ip=0)
decoder = Decoder(bitness, search_bytes, ip=jump_address)

for instruction in decoder:
# NOTE: this is not a None check, Instruction has special bool() handling
Expand All @@ -377,11 +377,10 @@ def get_hook_tail(self, jump_address: int) -> tuple[list[Instruction], int]:
f"Original code contains a far branch: {instruction}"
)

# TODO: try and fix the jump instead
if not near_target < self._jump_needed - position:
raise ValueError(
f"Original code contains a near jump outside of captured code: {instruction}"
)
# Near branches are handled by BlockEncoder:
# - Branches to instructions within this set: automatically linked
# - Branches outside this set: encoded to the original absolute target
# (safe because the original code at that address is not modified)

# TODO: figure out how xbegin works
case FlowControl.XBEGIN_XABORT_XEND:
Expand Down
4 changes: 3 additions & 1 deletion memobj/process/__init__.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
from .base import Process
from .module import Module
from .linux.module import LinuxModule
from .linux.process import LinuxProcess
from .windows.module import WindowsModule
from .windows.process import WindowsProcess


__all__ = ["Process", "Module", "WindowsModule", "WindowsProcess"]
__all__ = ["Process", "Module", "LinuxModule", "LinuxProcess", "WindowsModule", "WindowsProcess"]
Empty file.
127 changes: 127 additions & 0 deletions memobj/process/linux/module.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,127 @@
from __future__ import annotations

import subprocess
from pathlib import Path
from typing import TYPE_CHECKING, Self

from memobj.process.module import Module

if TYPE_CHECKING:
from .process import LinuxProcess


class LinuxModule(Module):
_symbols: dict[str, int] | None = None

@classmethod
def from_name(
cls, process: LinuxProcess, name: str, *, ignore_case: bool = True
) -> Self:
if ignore_case:
name = name.lower()

for module in cls.get_all_modules(process):
compare = module.name.lower() if ignore_case else module.name
if compare == name:
return module

raise ValueError(f"No module named {name!r}")

@classmethod
def get_all_modules(cls, process: LinuxProcess) -> list[Self]:
"""Build module list from /proc/{pid}/maps, merging segments of the same file."""
seen: dict[str, tuple[int, int]] = {} # path -> (min_base, max_end)

with open(f"/proc/{process.process_id}/maps") as f:
for line in f:
parts = line.split()
if len(parts) < 6:
continue

pathname = parts[5]
if not pathname or pathname.startswith("["):
continue

start_str, end_str = parts[0].split("-")
start = int(start_str, 16)
end = int(end_str, 16)

if pathname in seen:
existing_base, existing_end = seen[pathname]
seen[pathname] = (min(existing_base, start), max(existing_end, end))
else:
seen[pathname] = (start, end)

return [
cls(
name=Path(path).name,
base_address=base,
executable_path=path,
size=end - base,
process=process,
)
for path, (base, end) in seen.items()
]

def _is_position_independent(self) -> bool:
"""Return True for PIE executables and shared libraries (ELF type ET_DYN)."""
try:
with open(self.executable_path, "rb") as f:
header = f.read(18)
if len(header) < 18 or header[:4] != b"\x7fELF":
return False
endian = "little" if header[5] == 1 else "big"
e_type = int.from_bytes(header[16:18], endian)
return e_type == 3 # ET_DYN
except OSError:
return False

def get_symbols(self) -> dict[str, int]:
if self._symbols is not None:
return self._symbols

is_pie = self._is_position_independent()
symbols: dict[str, int] = {}

# Try both dynamic (-D) and static symbol tables; merge, with static
# taking precedence for address accuracy on plain executables.
for nm_args in (
["nm", "--defined-only", "-P", self.executable_path],
["nm", "-D", "--defined-only", "-P", self.executable_path],
):
try:
result = subprocess.run(
nm_args,
capture_output=True,
text=True,
check=False,
)
except FileNotFoundError:
break

for line in result.stdout.splitlines():
# nm -P format: name type value size
parts = line.split()
if len(parts) < 3:
continue
sym_name = parts[0].split("@")[0] # strip GLIBC version suffixes
sym_type = parts[1]
if sym_type in ("U", "w", "v"):
continue
try:
offset = int(parts[2], 16)
# For PIE binaries and shared libs, nm gives load-relative offsets
addr = self.base_address + offset if is_pie else offset
if sym_name not in symbols:
symbols[sym_name] = addr
except ValueError:
continue

self._symbols = symbols
return symbols

def get_symbol_with_name(self, name: str) -> int:
try:
return self.get_symbols()[name]
except KeyError:
raise ValueError(f"No symbol named {name!r}")
Loading