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
|
#!/usr/bin/env python3
"""s01 - reply consumer: did the server consume the command, and what came back?
Real scenario (Handoff-vGPU-IO.md, PART 91 + ADDENDA 2-7, 2026-10-01):
the physical backend booted the real GSP-RM and sent the pre-boot
``GSP_SET_SYSTEM_INFO`` (fn72) / ``SET_REGISTRY`` (fn73) RPCs; the only reply
the host could see was a generic ``GSP_INIT_DONE rpc_result=0x59``
(``NV_ERR_OPERATING_SYSTEM``) wrapped by ``RmStartup``. Cycles 2-13
(~11:40-16:52, ten supervised hardware cycles) were spent chasing which
collector step consumed which field because the RM log ring was never mapped
(LOGRM id bug) and the failure was un-journaled. Static/Ghidra-only work
(REPORT3-10) produced hypotheses (display, devid, chipset, ACPI) that needed a
live ``0x59``/success reply to accept or reject.
Analog: hook the demo dispatcher, watch the control block's ``result`` field,
post known/unknown/fault commands and decode the exact reply per command.
"""
from __future__ import annotations
import os
try: # package import (tests) or direct script execution (run_all)
from . import _common as C
except ImportError:
import sys as _sys
_sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import _common as C
def run() -> dict:
import greeneye as ge
backend = C.choose_backend("native")
out = C.artifact_dir("s01_reply_consumer")
commands = [
'ge.launch(kernel=DEMO_ELF, backend="%s")' % backend,
's.hook("rpc_dispatch", on_enter=record cmd/args)',
"s.watch(gsp_ctrl+0x40, size=8, kind='w') # result",
"s.memory.write_u64(ctrl+0x20, cmd); s.run(duration=...) # mailbox post",
"s.memory.read_u64(ctrl+0x40) # decode reply",
]
scenario = C.scenario(
ids=[
"Handoff-vGPU-IO.md PART 91 (2026-10-01) cycles 2-3",
"PART 91 ADDENDUM 2-7 (2026-10-01): NOCAT, fn72/fn73, LOGRM id, identity",
],
baseline=(
"cycles 2-13, ~5 h wall-clock and 10+ supervised hardware boots; "
"each candidate fix cost one full boot cycle (~5-6 min) and the "
"only visible reply was a generic 0x59"
),
projection=(
"real GSP-RM reply consumption is blocked on the H2 device model; "
"on the analog the same observe->decode step is one hook + one "
"watchpoint per command"
),
blocked=True,
device="H2.2 QEMU device model (RM message loop / real .fwimage)",
)
C.render_banner(
"s01_reply_consumer",
"When the host posts a command, did the dispatcher consume it and what exact reply came back?",
"0x59 RM-init chase: generic reply hid which collector step consumed which field",
scenario["static_baseline"],
commands,
)
cases = [
{"cmd": 1, "arg0": 0x1234, "label": "PING"},
{"cmd": 7, "arg0": 0, "label": "unknown-id"},
{"cmd": 6, "arg0": 0, "label": "FAULT"},
]
replies = []
session = ge.launch(kernel=C.DEMO_ELF, backend=backend, name="vd-s01")
try:
ctrl = session.symbols.lookup("gsp_ctrl")
dispatches = []
stores = []
session.hook(
"rpc_dispatch",
on_enter=lambda hit: dispatches.append(
{"cmd": hit.arg(0), "arg0": hit.arg(1), "arg1": hit.arg(2), "arg2": hit.arg(3)}
),
)
session.watch(ctrl + C.CTRL_RESULT, size=8, kind="w", callback=lambda hit: stores.append(hit.pc))
session.run(duration=0.3)
for case in cases:
before_count = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT)
session.memory.write_u64(ctrl + C.CTRL_ARG0, case["arg0"])
session.memory.write_u64(ctrl + C.CTRL_MAILBOX, case["cmd"])
session.run(duration=0.25)
after_count = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT)
result = session.memory.read_u64(ctrl + C.CTRL_RESULT)
expected = {
1: (0x1234 ^ C.PING_XOR),
7: (~7) & 0xFFFFFFFFFFFFFFFF,
6: None,
}[case["cmd"]]
replies.append(
{
"label": case["label"],
"cmd": case["cmd"],
"arg0": case["arg0"],
"consumed": after_count > before_count,
"result": result,
"expected": expected,
"result_ok": (expected is None) or (result == expected),
"rpc_count_delta": after_count - before_count,
}
)
if case["cmd"] == 6:
# ebreak trap: the dispatcher's post-handler stores never run.
session.step(1)
known = replies[0]
unknown = replies[1]
fault = replies[2]
payload = {
"success": known["result_ok"] and unknown["result_ok"] and not fault["consumed"],
"backend": backend,
"firmware": os.path.relpath(C.DEMO_ELF, C.TOOL_ROOT),
"firmware_sha256": C.sha256_file(C.DEMO_ELF),
"dispatches": dispatches,
"reply_store_pcs": stores,
"replies": replies,
"question": "consumed and decoded every posted command",
"answer": (
f"PING consumed, reply {known['result']:#x} == arg0 ^ 0xbeef "
f"(expected {known['expected']:#x}); unknown id returns ~cmd "
f"({unknown['result']:#x}); FAULT consumed={fault['consumed']} "
f"(trap before the reply store) - the exact first-error visibility "
"gap the 0x59 chase lacked"
),
"success_criteria": {
"pass": "3/3 dispatches observed; PING/unknown replies match the ABI; FAULT traps before result",
"fail": "a posted command is not consumed or a reply disagrees with the ABI",
},
"real_case": scenario,
}
C.write_json(os.path.join(out, "reply.json"), payload)
finally:
session.close()
return payload
def main() -> int:
return 0 if C.timed_demo(run, C.artifact_dir("s01_reply_consumer"))["success"] else 1
if __name__ == "__main__":
raise SystemExit(main())
|
