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
|
#!/usr/bin/env python3
"""s08 - mutation triage: mutate one field, classify the outcome.
Real scenario (docs/18-HARNESS-AUTOMATION.md R5 and
Handoff-vGPU-IO.md PART 77 / docs/19 3.4, 2026-09-30..2026-10-02): the
error-path corpus (sentinel/boundary pointers on ECHO/SUM, invalid command
IDs) is how the emulation's fail-closed rules are derived. Statically the
demo's 4-of-24 outcomes were invisible; the discovery sweep found
``rpc_echo``/``rpc_sum`` fault on sentinel/boundary pointers while
``rpc_ping``/``rpc_state``/``rpc_fill`` stay on the handler path.
Analog: a seeded mutation table over command/arguments, one fresh session per
attempt (the documented safety strategy), classified as ok / trap / fault /
no-progress from stop signals, ``rpc_count`` and the result field.
"""
from __future__ import annotations
import json
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
FAULT_SIGNALS = {4, 7, 11} # SIGILL, SIGBUS, SIGSEGV
TRAP_SIGNAL = 5 # SIGTRAP (ebreak)
def _probe(ge, mutation: dict, backend: str) -> dict:
session = ge.launch(kernel=C.DEMO_ELF, backend=backend, name="vd-s08")
try:
ctrl = session.symbols.lookup("gsp_ctrl")
signals = []
session.on_event("stop", lambda hit, sig=signals: sig.append(hit.stop.signal) if hit.stop else None)
session.run(duration=0.15)
literal = 0
if mutation.get("arg0") == "hello":
hits = session.memory.search(0x80000000, 0x2000, b"hello-green-eye")
literal = hits[0] if hits else ctrl + C.CTRL_BUF
before = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT)
session.memory.write_u64(ctrl + C.CTRL_ARG0, literal if mutation.get("arg0") == "hello" else mutation.get("arg0", 0))
session.memory.write_u64(ctrl + C.CTRL_ARG1, mutation.get("arg1", 0))
session.memory.write_u64(ctrl + C.CTRL_MAILBOX, mutation["cmd"])
# max_stops=1: a faulted target keeps re-faulting at the same PC, so
# stop at the first signal instead of spinning the fault loop.
session.run(duration=0.2, max_stops=1)
after = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT)
if after == before and not any(signals):
session.run(duration=0.3, max_stops=1)
after = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT)
result = session.memory.read_u64(ctrl + C.CTRL_RESULT)
if any(s == TRAP_SIGNAL for s in signals):
outcome = "trap"
elif any(s in FAULT_SIGNALS for s in signals):
outcome = "fault"
elif after > before:
outcome = "ok"
else:
outcome = "no-progress"
return {
"outcome": outcome,
"rpc_count_delta": after - before,
"result": result,
"result_hex": f"{result:#x}",
"signals": [s for s in signals if s],
"consumed": after > before,
}
finally:
session.close()
def run() -> dict:
import greeneye as ge
out = C.artifact_dir("s08_mutation_triage")
backend = C.choose_backend("native")
mutations = [
{"strategy": "arg-boundary", "cmd": 2, "arg0": 0, "arg1": 0, "expect": "fault"},
{"strategy": "valid", "cmd": 2, "arg0": "hello", "arg1": 0, "expect": "ok"},
{"strategy": "arg-boundary", "cmd": 2, "arg0": 0xFFFFFFFFFFFFFFFF, "arg1": 0, "expect": "fault"},
{"strategy": "valid", "cmd": 4, "arg0": "hello", "arg1": 5, "expect": "ok"},
{"strategy": "buffer-length", "cmd": 4, "arg0": "hello", "arg1": 0x8000000000000000, "expect": "no-progress"},
{"strategy": "buffer-length", "cmd": 3, "arg0": 0x41, "arg1": 4096, "expect": "ok"},
{"strategy": "cmd-invalid", "cmd": 7, "arg0": 0, "arg1": 0, "expect": "ok"},
]
commands = [
"for each seeded mutation: fresh native session, settle, one mailbox write",
"run bounded; classify from stop signals + rpc_count + result",
"write findings.jsonl (deterministic order) for corpus diffing",
]
scenario = C.scenario(
ids=[
"docs/18-HARNESS-AUTOMATION.md 2 (R5): bounded seeded mutation, outcome classes",
"docs/19-CONTROL-DISCOVERY.md 3.4 (2026-10-02): sentinel/boundary pointer faults found by the sweep",
"Handoff-vGPU-IO.md PARTs 3-9 (2026-09-21..23): hand-derived assert/error paths",
],
baseline=(
"statically the fault surface is invisible; the project derived error "
"paths by hand from asserts and NOCAT records across many physical "
"cycles; a 64-iteration QEMU mutation run is the tracked baseline"
),
projection=(
"real RM controls land in the same mutation harness once the RM "
"surface exists (docs/19 section 5); one device-write experiment "
"per physical cycle is the current cap"
),
blocked=True,
device="H1.3 RM control surface for mutation of real controls",
)
C.render_banner(
"s08_mutation_triage",
"Mutate one input field deterministically and classify the outcome (ok/trap/fault/no-progress).",
"error-path corpus for fail-closed emulation rules (R5/H4.3)",
scenario["static_baseline"],
commands,
)
findings = []
for index, mutation in enumerate(mutations):
observed = _probe(ge, mutation, backend)
finding = {"iteration": index, **mutation, **observed, "matched_expectation": observed["outcome"] == mutation["expect"]}
findings.append(finding)
classes: dict[str, int] = {}
for finding in findings:
classes[finding["outcome"]] = classes.get(finding["outcome"], 0) + 1
payload = {
"success": all(f["matched_expectation"] for f in findings),
"backend": backend,
"mutations": len(findings),
"outcome_classes": classes,
"findings": findings,
"question": "enumerate and classify one-field mutations without hardware",
"answer": (
f"{len(findings)} mutations -> {classes}; all expectations matched; "
"faults localised to the dereferencing handlers, the oversized SUM "
"stays no-progress, FILL clamps at the buffer size - a corpus entry "
"per mutation with no device writes"
),
"success_criteria": {
"pass": "every seeded mutation produces its documented outcome class",
"fail": "an outcome contradicts the case table or a session leaks",
},
"real_case": scenario,
}
with open(os.path.join(out, "findings.jsonl"), "w", encoding="utf-8") as fh:
for finding in findings:
fh.write(json.dumps(finding, sort_keys=True) + "\n")
C.write_json(os.path.join(out, "mutation.json"), payload)
return payload
def main() -> int:
return 0 if C.timed_demo(run, C.artifact_dir("s08_mutation_triage"))["success"] else 1
if __name__ == "__main__":
raise SystemExit(main())
|
