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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
|
"""Frida-like service injection for risc-v-gsp-rm-debugger.
Loads a position-independent RV64 service blob into a live target, registers
its dispatch entry with the firmware (on the demo uproc: a handler-table slot
that commands 16..23 dispatch to), and invokes it either directly through
``Session.call`` (a debugger-side remote function call) or through the
firmware's own mailbox dispatcher.
On a future GSP-RM simulator the same shape becomes "inject an RM engine and
register it with the simulated RM core"; see ``docs/12-INJECTION-SDK.md`` and
``RISC-V-GSP-RM-DEBUGGER-ROADMAP.md``.
"""
from __future__ import annotations
from dataclasses import dataclass
from .elf import load_elf
from .errors import GreenEyeError
#: Symbol the injector calls in a service blob.
ENTRY_SYMBOL = "ge_demo_service_dispatch"
#: Session attribute holding the per-session service registry (load order).
SERVICE_REGISTRY_ATTR = "_injected_services"
#: Demo uproc control-block field offsets (see firmware/uproc_demo).
CTRL_HANDLERS = 0x150
CTRL_MAILBOX = 0x20
CTRL_ARG0 = 0x28
CTRL_ARG1 = 0x30
CTRL_ARG2 = 0x38
CTRL_RESULT = 0x40
INJECT_CMD_BASE = 16
@dataclass
class InjectedService:
path: str
name: str
load_addr: int
entry_addr: int
dispatch_offset: int
slot: int
size: int
def summary(self) -> str:
return (
f"{self.name}: {self.size} bytes at {self.load_addr:#x}, "
f"entry {self.entry_addr:#x} (offset {self.dispatch_offset:#x}), slot {self.slot}"
)
def to_dict(self) -> dict:
"""Plain-data view for MCP/reporting (no behavior)."""
return {
"path": self.path,
"name": self.name,
"load_addr": self.load_addr,
"entry_addr": self.entry_addr,
"dispatch_offset": self.dispatch_offset,
"slot": self.slot,
"size": self.size,
}
def services(session) -> list["InjectedService"]:
"""Services loaded into ``session`` this process, in load order.
The registry lives on the session object (``SERVICE_REGISTRY_ATTR``) and is
populated by :func:`load_service`; it is process-local and not part of the
on-disk session format.
"""
return list(getattr(session, SERVICE_REGISTRY_ATTR, ()) or ())
def _register(session, service: InjectedService) -> None:
registry = getattr(session, SERVICE_REGISTRY_ATTR, None)
if registry is None:
registry = []
setattr(session, SERVICE_REGISTRY_ATTR, registry)
registry.append(service)
def load_service(
session,
service_elf: str,
*,
load_addr: int | None = None,
slot: int = 6,
entry_symbol: str = ENTRY_SYMBOL,
) -> InjectedService:
"""Load a service blob and register it with the demo firmware."""
image = load_elf(service_elf)
entry = None
for symbol in image.symbols:
if symbol.name == entry_symbol:
entry = symbol
break
if entry is None:
available = ", ".join(sorted(s.name for s in image.symbols if s.is_function)) or "<none>"
raise GreenEyeError(f"service {service_elf}: symbol {entry_symbol!r} not found (functions: {available})")
if not 0 <= slot <= 7:
raise GreenEyeError(f"slot must be 0..7 (got {slot})")
if load_addr is None:
base = getattr(session.target, "ram_base", 0x80000000) if session.target else 0x80000000
size = getattr(session.target, "memory_mb", 256) * 1024 * 1024 if session.target else 256 * 1024 * 1024
load_addr = base + 0x01000000
if load_addr + image.max_vaddr + 0x1000 >= base + size:
load_addr = base + max(0x1000, size - image.max_vaddr - 0x2000)
if getattr(session.target, "memory_mb", 0):
ram_end = getattr(session.target, "ram_base", 0x80000000) + session.target.memory_mb * 1024 * 1024
if load_addr + image.max_vaddr > ram_end:
raise GreenEyeError(
f"service does not fit: load {load_addr:#x} + {image.max_vaddr:#x} exceeds RAM end {ram_end:#x}"
)
total = 0
for segment in image.load_segments:
data = image.segment_bytes(segment)
if data:
session.memory.write(load_addr + segment.p_vaddr, data)
total += len(data)
entry_addr = load_addr + entry.value
ctrl = session.symbols.get("gsp_ctrl")
if ctrl is not None:
session.memory.write_u64(ctrl.addr + CTRL_HANDLERS + slot * 8, entry_addr)
service = InjectedService(
path=service_elf,
name=service_elf.rsplit("/", 1)[-1],
load_addr=load_addr,
entry_addr=entry_addr,
dispatch_offset=entry.value,
slot=slot,
size=total,
)
session.recorder.event(
"service_injected",
path=service.path,
name=service.name,
load_addr=load_addr,
entry_addr=entry_addr,
slot=slot,
size=total,
)
_register(session, service)
return service
def invoke_remote(session, service: InjectedService, cmd: int, a0: int = 0, a1: int = 0, timeout: float = 3.0) -> int:
"""Call the service directly from the debugger (Frida NativeFunction)."""
return session.call(service.entry_addr, (cmd, a0, a1), timeout=timeout)
def invoke_via_firmware(session, service: InjectedService, cmd: int, a0: int = 0, a1: int = 0, settle: float = 0.25) -> int:
"""Invoke the service through the firmware's mailbox dispatcher."""
ctrl = session.symbols.lookup("gsp_ctrl")
session.memory.write_u64(ctrl + CTRL_ARG0, cmd)
session.memory.write_u64(ctrl + CTRL_ARG1, a0)
session.memory.write_u64(ctrl + CTRL_ARG2, a1)
session.memory.write_u64(ctrl + CTRL_MAILBOX, INJECT_CMD_BASE + service.slot)
session.run(duration=settle)
return session.memory.read_u64(ctrl + CTRL_RESULT)
|
