GP-5637: More improvements to the PyGhidra API

This commit is contained in:
Ryan Kurtz 2025-09-12 08:29:09 -04:00
parent f52f5a7695
commit 26685abc75
3 changed files with 15 additions and 15 deletions

View file

@ -41,10 +41,16 @@ public class PyGhidraTaskMonitor implements TaskMonitor {
private Timer timer = new Timer(); private Timer timer = new Timer();
private WeakSet<CancelledListener> listeners = private WeakSet<CancelledListener> listeners =
WeakDataStructureFactory.createCopyOnReadWeakSet(); WeakDataStructureFactory.createCopyOnReadWeakSet();
private TriConsumer<String, Long, Long> changeCallback; private TriConsumer<String, Long, Long> changeCallback;
/**
* Creates a new {@link PyGhidraTaskMonitor}
*
* @param timeoutSecs The number of seconds before a cancellation timeout is triggered, or
* {@code null} for no timeout
* @param changeCallback A function that gets called any time a change to the monitor occurred,
* or {@code null} for no callback
*/
public PyGhidraTaskMonitor(Integer timeoutSecs, public PyGhidraTaskMonitor(Integer timeoutSecs,
TriConsumer<String, Long, Long> changeCallback) { TriConsumer<String, Long, Long> changeCallback) {
isCancelled = false; isCancelled = false;

View file

@ -223,15 +223,12 @@ def program_loader() -> "ProgramLoader.Builder":
### pyghidra.monitor() ### pyghidra.monitor()
```python ```python
def monitor( def monitor(
timeout: Optional[int] = None, timeout: Optional[int] = None
change_callback: Callable[[str, int, int], None] = None
) -> "PyGhidraTaskMonitor": ) -> "PyGhidraTaskMonitor":
""" """
Convenience function to get a "PyGhidraTaskMonitor" object. Convenience function to get a "PyGhidraTaskMonitor" object.
:param timeout: An optional number of seconds to wait before canceling the monitor. :param timeout: An optional number of seconds to wait before canceling the monitor.
:param change_callback: A optional function that gets called any time the monitor receives an
update.
:return: A "PyGhidraTaskMonitor" object. :return: A "PyGhidraTaskMonitor" object.
""" """
``` ```
@ -289,7 +286,7 @@ with pyghidra.open_project(os.environ["GHIDRA_PROJECT_DIR"], "ExampleProject", c
with pyghidra.open_filesystem(f"{os.environ['DOWNLOADS_DIR']}/ghidra_11.4_PUBLIC_20250620.zip") as fs: with pyghidra.open_filesystem(f"{os.environ['DOWNLOADS_DIR']}/ghidra_11.4_PUBLIC_20250620.zip") as fs:
loader = pyghidra.program_loader().project(project) loader = pyghidra.program_loader().project(project)
for f in fs.files(lambda f: "os/" in f.path and f.name.startswith("decompile")): for f in fs.files(lambda f: "os/" in f.path and f.name.startswith("decompile")):
loader.source(f.getFSRL()).projectFolderPath("/" + f.parentFile.name) loader = loader.source(f.getFSRL()).projectFolderPath("/" + f.parentFile.name)
with loader.load() as load_results: with loader.load() as load_results:
load_results.save(pyghidra.monitor()) load_results.save(pyghidra.monitor())
@ -298,10 +295,10 @@ with pyghidra.open_project(os.environ["GHIDRA_PROJECT_DIR"], "ExampleProject", c
analysis_props = pyghidra.analysis_properties(program) analysis_props = pyghidra.analysis_properties(program)
with pyghidra.transaction(program): with pyghidra.transaction(program):
analysis_props.setBoolean("Non-Returning Functions - Discovered", False) analysis_props.setBoolean("Non-Returning Functions - Discovered", False)
pyghidra.analyze(program, pyghidra.monitor(10)) analysis_log = pyghidra.analyze(program, pyghidra.monitor(10))
program.save("Analyzed", pyghidra.monitor()) program.save("Analyzed", pyghidra.monitor())
# Walk the project and set a propery in each decompiler program # Walk the project and set a property in each decompiler program
def set_property(domain_file, program): def set_property(domain_file, program):
with pyghidra.transaction(program): with pyghidra.transaction(program):
program_info = pyghidra.program_info(program) program_info = pyghidra.program_info(program)
@ -313,7 +310,7 @@ with pyghidra.open_project(os.environ["GHIDRA_PROJECT_DIR"], "ExampleProject", c
ByteArrayCls = jpype.JArray(jpype.JByte) ByteArrayCls = jpype.JArray(jpype.JByte)
my_bytes = ByteArrayCls(b"\xaa\xbb\xcc\xdd\xee\xff") my_bytes = ByteArrayCls(b"\xaa\xbb\xcc\xdd\xee\xff")
loader = pyghidra.program_loader().project(project).source(my_bytes).name("my_bytes") loader = pyghidra.program_loader().project(project).source(my_bytes).name("my_bytes")
loader.loaders("BinaryLoader").language("DATA:LE:64:default") loader = loader.loaders("BinaryLoader").language("DATA:LE:64:default")
with loader.load() as load_results: with loader.load() as load_results:
load_results.save(pyghidra.monitor()) load_results.save(pyghidra.monitor())

View file

@ -301,21 +301,18 @@ def program_loader() -> "ProgramLoader.Builder":
return ProgramLoader.builder() return ProgramLoader.builder()
def monitor( def monitor(
timeout: Optional[int] = None, timeout: Optional[int] = None
change_callback: Callable[[str, int, int], None] = None
) -> "PyGhidraTaskMonitor": ) -> "PyGhidraTaskMonitor":
""" """
Convenience function to get a "PyGhidraTaskMonitor" object. Convenience function to get a "PyGhidraTaskMonitor" object.
:param timeout: An optional number of seconds to wait before canceling the monitor. :param timeout: An optional number of seconds to wait before canceling the monitor.
:param change_callback: A optional function that gets called any time the monitor receives an
update.
:return: A "PyGhidraTaskMonitor" object. :return: A "PyGhidraTaskMonitor" object.
""" """
from ghidra.pyghidra import PyGhidraTaskMonitor from ghidra.pyghidra import PyGhidraTaskMonitor
from jpype.types import JInt from jpype.types import JInt
t = None if timeout is None else JInt(timeout) t = None if timeout is None else JInt(timeout)
return PyGhidraTaskMonitor(t, change_callback) return PyGhidraTaskMonitor(t, None)
def walk_project( def walk_project(
project: "Project", project: "Project",