1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
|
"""green-eye plugin system.
A plugin is a normal Python module that exposes::
def register(ge):
ge.log("my plugin loaded")
ge.add_command("mycmd", run_mycmd, help="…")
ge.hook("rpc_sum", on_enter=lambda hit: ge.log(f"sum({hit.arg(0)})"))
Plugins can extend green-eye with:
* new CLI commands (including an MCP server later - roadmap);
* runtime RE tools (extra watchpoint strategies, structure probes, coverage);
* virtualisation-environment updates (e.g. record/replay harnesses);
* AI-harness integration and reporting.
Discovery order:
1. explicit ``--plugin PATH`` arguments;
2. colon-separated paths in ``GREENEYE_PLUGINS``;
3. ``<tool>/plugins/*.py`` (the bundled examples).
"""
from __future__ import annotations
import importlib.util
import os
import sys
from typing import Callable
class GreenEyeAPI:
"""Handle passed to ``register(ge)``."""
def __init__(self, session, cli=None):
self.session = session
self.cli = cli
self.commands: dict[str, tuple[Callable, str]] = {}
self.loaded: list[str] = []
# -- convenience --------------------------------------------------------
@property
def memory(self):
return self.session.memory
@property
def symbols(self):
return self.session.symbols
@property
def recorder(self):
return self.session.recorder
def log(self, message: str, **fields) -> None:
self.session.recorder.event("plugin", message=message, **fields)
def hook(self, target, on_enter=None, on_leave=None, **kw):
return self.session.hook(target, on_enter=on_enter, on_leave=on_leave, **kw)
def watch(self, target, size: int = 8, kind: str = "w", callback=None, **kw):
return self.session.watch(target, size=size, kind=kind, callback=callback, **kw)
def on_event(self, kind: str, callback) -> None:
self.session.on_event(kind, callback)
def resolve(self, target):
return self.session.resolve(target)
def add_command(self, name: str, fn: Callable, help: str = "") -> None:
self.commands[name] = (fn, help)
def run_command(self, name: str, args: list[str]) -> int:
fn, _help = self.commands[name]
return int(fn(self.session, args) or 0)
class PluginManager:
def __init__(self, session, cli=None):
self.session = session
self.api = GreenEyeAPI(session, cli=cli)
self.loaded: list[str] = []
def default_paths(self) -> list[str]:
here = os.path.dirname(os.path.abspath(__file__))
bundled = os.path.join(os.path.dirname(here), "plugins")
paths = []
env = os.environ.get("GREENEYE_PLUGINS", "")
paths += [p for p in env.split(os.pathsep) if p]
if os.path.isdir(bundled):
paths.append(bundled)
return paths
def load_all(self, explicit: list[str] | None = None) -> list[str]:
for path in (explicit or []) + self.default_paths():
if os.path.isdir(path):
for entry in sorted(os.listdir(path)):
if entry.endswith(".py") and not entry.startswith("_"):
self._load_file(os.path.join(path, entry))
elif path.endswith(".py") and os.path.isfile(path):
self._load_file(path)
return self.loaded
def _load_file(self, path: str) -> None:
name = os.path.splitext(os.path.basename(path))[0]
mod_name = f"greeneye_plugin_{name}"
try:
spec = importlib.util.spec_from_file_location(mod_name, path)
if spec is None or spec.loader is None:
return
module = importlib.util.module_from_spec(spec)
sys.modules[mod_name] = module
spec.loader.exec_module(module)
register = getattr(module, "register", None)
if register is None:
self.session.recorder.event("plugin_error", plugin=name, error="no register(ge)")
return
register(self.api)
self.loaded.append(name)
self.session.recorder.event("plugin_loaded", plugin=name)
except Exception as exc: # plugins must never break the session
self.session.recorder.event("plugin_error", plugin=name, error=repr(exc))
|
