GP-6273: post-review

GP-6273: working
This commit is contained in:
d-millar
2026-01-09 13:09:53 -05:00
parent f9cee4d639
commit bcced66f18
6 changed files with 107 additions and 23 deletions
@@ -38,7 +38,7 @@
<UL style="list-style-type: none">
<LI>
<PRE>
python3 -m pip install psutil protobuf>=3.20.3
python3 -m pip install protobuf>=3.20.3
</PRE>
</LI>
</UL>
@@ -51,7 +51,7 @@ python3 -m pip install psutil protobuf>=3.20.3
<PRE>
cd /path/to/ghidra_<EM>
version</EM>/Ghidra/Debug
python3 -m pip install --no-index -f Debugger-rmi-trace/pypkg/dist -f Debugger-agent-gdb/pypkg/dist psutil protobuf
python3 -m pip install --no-index -f Debugger-rmi-trace/pypkg/dist -f Debugger-agent-gdb/pypkg/dist protobuf
</PRE>
</LI>
</UL>
@@ -22,11 +22,6 @@ import time
from typing import (Any, Callable, Dict, Generator, List, Optional, Sequence,
Tuple, Type, TypeVar, Union)
try:
import psutil
except ImportError:
print(f"Unable to import 'psutil' - check that it has been installed")
from ghidratrace import sch
from ghidratrace.client import (Client, Address, AddressRange, Lifespan,
Schedule, Trace, TraceObject, TraceObjectValue,
@@ -1070,18 +1065,21 @@ def ghidra_trace_put_inferiors(*, is_mi: bool, **kwargs) -> None:
put_inferiors()
def put_available() -> None:
# TODO: Compared to -list-thread-groups --available:
# Is that always from the host, or can that pslist a remote target?
# psutil will always be from the host.
def put_available(availables: Optional[List[util.Available]] = None) -> List[util.Available]:
trace = STATE.require_trace()
inf = gdb.selected_inferior()
if availables is None:
try:
availables = util.AVAILABLE_INFO_READER.get_availables()
except Exception:
availables = []
keys = []
for proc in psutil.process_iter():
for proc in availables:
ppath = AVAILABLE_PATTERN.format(pid=proc.pid)
procobj = trace.create_object(ppath)
keys.append(AVAILABLE_KEY_PATTERN.format(pid=proc.pid))
procobj.set_value('PID', proc.pid)
procobj.set_value('_display', f'{proc.pid} {proc.name()}')
procobj.set_value('_display', f'{proc.pid} {proc.command}')
procobj.insert()
trace.proxy_object_path(AVAILABLES_PATH).retain_values(keys)
@@ -90,6 +90,13 @@ class Section:
return Section(self.name, start, end, offset, list(attrs))
@dataclass(frozen=True)
class Available:
pid: int
user: str
command: str
@dataclass(frozen=True)
class Module:
name: str
@@ -296,6 +303,54 @@ def _choose_region_info_reader() -> RegionInfoReader:
REGION_INFO_READER = _choose_region_info_reader()
AVAILABLES_CMD = 'info os processes'
AVAILABLE_PATTERN = re.compile("\\s*" +
"(?P<pid>[0-9]+)\\s+" +
"(?P<user>[0-9,A-Z,a-z]+)\\s+" +
"(?P<command>.*)")
class AvailableInfoReader(object):
cmd = AVAILABLES_CMD
available_pattern = AVAILABLE_PATTERN
def available_from_line(self, line: str, max_addr: int) -> Optional[Available]:
mat = self.available_pattern.fullmatch(line)
if mat is None:
return None
pid = mat['pid']
user = mat['user']
command = mat['command']
return Available(pid, user, command)
def get_availables(self) -> List[Available]:
availables: List[Available] = []
try:
out = gdb.execute(self.cmd, to_string=True)
max_addr = compute_max_addr()
except:
return availables
for line in out.split('\n'):
a = self.available_from_line(line, max_addr)
if a is None:
continue
availables.append(a)
return availables
def have_changed(self, availables: List[Available]) -> Tuple[bool, Optional[List[Available]]]:
new_availables = self.get_availables()
if new_availables == availables and len(new_availables) > 0:
return False, None
return True, new_availables
def _choose_available_info_reader() -> AvailableInfoReader:
return AvailableInfoReader()
AVAILABLE_INFO_READER = _choose_available_info_reader()
BREAK_LOCS_CMD = 'info break {}'
BREAK_PATTERN = re.compile('')
BREAK_LOC_PATTERN = re.compile('')
@@ -38,7 +38,7 @@
<UL style="list-style-type: none">
<LI>
<PRE>
python3 -m pip install psutil protobuf>=3.20.3
python3 -m pip install protobuf>=3.20.3
</PRE>
</LI>
</UL>
@@ -54,7 +54,7 @@ python3 -m pip install psutil protobuf>=3.20.3
<PRE>
cd /path/to/ghidra_<EM>
version</EM>/Ghidra/Debug
python3 -m pip install --no-index -f Debugger-rmi-trace/pypkg/dist -f Debugger-agent-lldb/pypkg/dist psutil protobuf
python3 -m pip install --no-index -f Debugger-rmi-trace/pypkg/dist -f Debugger-agent-lldb/pypkg/dist protobuf
</PRE>
</LI>
</UL>
@@ -26,11 +26,6 @@ import time
from typing import (Any, Callable, Dict, Generator, List, Literal,
Optional, Tuple, TypeVar, Union, cast)
try:
import psutil
except ImportError:
print("Unable to import 'psutil' - check that it has been installed")
from ghidratrace import sch
from ghidratrace.client import (Client, Address, AddressRange, Trace, Schedule,
TraceObject, Transaction)
@@ -1478,15 +1473,17 @@ def ghidra_trace_put_processes(debugger: lldb.SBDebugger, command: str,
put_processes()
def put_available() -> None:
def put_available(availables: Optional[List[util.Available]] = None) -> List[util.Available]:
trace = STATE.require_trace()
availables = util.AVAILABLE_INFO_READER.get_availables()
keys = []
for proc in psutil.process_iter():
for proc in availables:
ppath = AVAILABLE_PATTERN.format(pid=proc.pid)
procobj = trace.create_object(ppath)
keys.append(AVAILABLE_KEY_PATTERN.format(pid=proc.pid))
procobj.set_value('PID', proc.pid)
procobj.set_value('_display', f'{proc.pid} {proc.name()}')
procobj.set_value('Name', proc.name)
procobj.set_value('_display', f'{proc.pid} {proc.command}')
procobj.insert()
trace.proxy_object_path(AVAILABLES_PATH).retain_values(keys)
@@ -178,6 +178,40 @@ def _choose_region_info_reader() -> RegionInfoReader:
REGION_INFO_READER = _choose_region_info_reader()
@dataclass
class Available:
pid: int
name: str
command: str
class AvailableInfoReader(object):
def available_from_sbprocinfo(self, info: lldb.SBProcessInfo) -> Available:
pid = info.GetProcessID()
name = info.GetName()
command = info.GetExecutableFile()
return Available(pid, name, command)
def get_availables(self) -> List[Available]:
availables = []
platform = get_debugger().GetPlatformAtIndex(0)
err = lldb.SBError()
proclist = platform.GetAllProcesses(err)
for i in range(0, proclist.GetSize()):
info = lldb.SBProcessInfo()
success = proclist.GetProcessInfoAtIndex(i, info)
if success:
a = self.available_from_sbprocinfo(info)
availables.append(a)
return availables
def _choose_available_info_reader() -> AvailableInfoReader:
return AvailableInfoReader()
AVAILABLE_INFO_READER = _choose_available_info_reader()
BREAK_LOCS_CMD = 'breakpoint list {}'
BREAK_PATTERN = re.compile('')
BREAK_LOC_PATTERN = re.compile('')