mirror of
https://github.com/morluto/rea.git
synced 2026-10-12 02:19:24 +00:00
fix(hopper): preserve owned basic-block endpoint instructions (#1118)
* fix(hopper): preserve owned basic-block endpoint instructions * test(hopper): share probe runner and restore fixture hashability Reuse one probe helper with the facade test's execFile options (encoding, timeout, maxBuffer) instead of repeating the invocation. Give the fixture Block a hash consistent with its equality so the equality-based ownership doubles stay hashable. * fix(hopper): align owned instruction walks with normalized block ranges --------- Co-authored-by: Vilmer Frost <[email protected]> Co-authored-by: morluto <[email protected]>
This commit is contained in:
vilmerfrost
co-authored by
Vilmer Frost
morluto
Vilmer Frost
morluto
parent
2b27515476
commit
329cf206d1
+32
-12
@@ -318,34 +318,54 @@ def _containing_procedure(document, address):
|
||||
|
||||
def _basic_block_end(procedure, block):
|
||||
"""Normalize native block endpoints to an exclusive byte address."""
|
||||
owner_at = getattr(procedure, "getBasicBlockAtAddress", None)
|
||||
if not callable(owner_at):
|
||||
raise CapabilityUnavailableError(
|
||||
"Instruction enumeration requires Hopper basic-block ownership"
|
||||
)
|
||||
end = block.getEndingAddress()
|
||||
# Hopper 6.1 returns the final instruction address, despite the Python
|
||||
# documentation describing an exclusive end. Membership distinguishes
|
||||
# that representation from an actual exclusive endpoint in other builds.
|
||||
if procedure.getBasicBlockAtAddress(end) == block:
|
||||
if owner_at(end) == block:
|
||||
instruction = procedure.getSegment().getInstructionAtAddress(end)
|
||||
if instruction is not None and instruction.getInstructionLength() > 0:
|
||||
return end + instruction.getInstructionLength()
|
||||
return end
|
||||
|
||||
|
||||
def _block_instruction_addresses(procedure, block):
|
||||
"""Walk either endpoint convention using Hopper's block ownership API.
|
||||
|
||||
Some Hopper builds return the final instruction start, despite the public
|
||||
API documenting an exclusive end. Equality compares fresh native wrappers;
|
||||
probing ownership also keeps an exclusive endpoint in the next block out.
|
||||
"""
|
||||
segment = procedure.getSegment()
|
||||
address = block.getStartingAddress()
|
||||
end = _basic_block_end(procedure, block)
|
||||
while address < end:
|
||||
if procedure.getBasicBlockAtAddress(address) != block:
|
||||
break
|
||||
instruction = segment.getInstructionAtAddress(address)
|
||||
if instruction is None:
|
||||
break
|
||||
length = instruction.getInstructionLength()
|
||||
if length <= 0 or address + length > end:
|
||||
break
|
||||
yield address
|
||||
address += length
|
||||
|
||||
|
||||
def _instruction_addresses(procedure):
|
||||
result = []
|
||||
seen = set()
|
||||
segment = procedure.getSegment()
|
||||
for block in procedure.basicBlockIterator():
|
||||
address = block.getStartingAddress()
|
||||
end = _basic_block_end(procedure, block)
|
||||
while address < end and address not in seen:
|
||||
for address in _block_instruction_addresses(procedure, block):
|
||||
if address in seen:
|
||||
continue
|
||||
seen.add(address)
|
||||
instruction = segment.getInstructionAtAddress(address)
|
||||
if instruction is None:
|
||||
break
|
||||
result.append(address)
|
||||
length = instruction.getInstructionLength()
|
||||
if length <= 0:
|
||||
break
|
||||
address += length
|
||||
return result
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,105 @@
|
||||
import { execFile } from "node:child_process";
|
||||
import { promisify } from "node:util";
|
||||
import { fileURLToPath } from "node:url";
|
||||
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { z } from "zod";
|
||||
|
||||
const execute = promisify(execFile);
|
||||
const bridge = fileURLToPath(
|
||||
new URL("../../../../bridge/hopper_bridge.py", import.meta.url),
|
||||
);
|
||||
const probe = fileURLToPath(
|
||||
new URL("../../../fixtures/hopperInstructionProbe.py", import.meta.url),
|
||||
);
|
||||
const resultSchema = z.object({
|
||||
addresses: z.array(z.number()),
|
||||
blocks: z.array(z.object({ start: z.string(), end: z.string() })),
|
||||
length: z.number(),
|
||||
assembly: z.array(z.string()),
|
||||
fast: z.array(z.string()),
|
||||
dossier: z.array(z.string()),
|
||||
references: z.array(z.object({ source_address: z.string() })),
|
||||
});
|
||||
|
||||
type ProbeResult = z.infer<typeof resultSchema>;
|
||||
|
||||
async function runProbe(scenario: string): Promise<ProbeResult> {
|
||||
const { stdout } = await execute("python3", [probe, bridge, scenario], {
|
||||
encoding: "utf8",
|
||||
timeout: 3_000,
|
||||
maxBuffer: 1_024 * 1_024,
|
||||
});
|
||||
return resultSchema.parse(JSON.parse(stdout));
|
||||
}
|
||||
|
||||
describe("Hopper instruction block boundaries", () => {
|
||||
it.each(["inclusive", "exclusive", "duplicate"])(
|
||||
"preserves branches and returns with %s endpoints without duplicates or adjacent instructions",
|
||||
async (scenario) => {
|
||||
const result = await runProbe(scenario);
|
||||
expect(result.addresses).toEqual([0x1000, 0x1004, 0x1008]);
|
||||
if (scenario !== "duplicate") {
|
||||
expect(result.blocks).toEqual([
|
||||
{ start: "0x1000", end: "0x1008" },
|
||||
{ start: "0x1008", end: "0x100c" },
|
||||
]);
|
||||
expect(result.length).toBe(12);
|
||||
}
|
||||
const lines = ["0x1000: mov", "0x1004: b.ne", "0x1008: ret"];
|
||||
expect(result.assembly).toEqual(lines);
|
||||
expect(result.fast).toEqual(lines);
|
||||
expect(result.dossier).toEqual(lines);
|
||||
expect(result.references.map((edge) => edge.source_address)).toEqual([
|
||||
"0x1004",
|
||||
]);
|
||||
},
|
||||
);
|
||||
|
||||
it.each([
|
||||
["gap", [0x1000]],
|
||||
["zero_length", [0x1000]],
|
||||
["negative_length", [0x1000]],
|
||||
["cross_boundary", []],
|
||||
["foreign_owner", [0x1000, 0x1004]],
|
||||
["reversed", []],
|
||||
])("stops safely at %s", async (scenario, addresses) => {
|
||||
const result = await runProbe(scenario);
|
||||
expect(result.addresses).toEqual(addresses);
|
||||
expect(result.assembly).toHaveLength(addresses.length);
|
||||
expect(result.fast).toEqual(result.assembly);
|
||||
expect(result.dossier).toEqual(result.assembly);
|
||||
});
|
||||
|
||||
it("uses instruction lengths without assuming ARM64 fixed-width encoding", async () => {
|
||||
const result = await runProbe("variable_length");
|
||||
expect(result.addresses).toEqual([0x1000, 0x1003, 0x1005]);
|
||||
expect(result.blocks).toEqual([
|
||||
{ start: "0x1000", end: "0x1005" },
|
||||
{ start: "0x1005", end: "0x1006" },
|
||||
]);
|
||||
expect(result.length).toBe(6);
|
||||
expect(result.assembly).toEqual([
|
||||
"0x1000: mov",
|
||||
"0x1003: b.ne",
|
||||
"0x1005: ret",
|
||||
]);
|
||||
expect(result.fast).toEqual(result.assembly);
|
||||
expect(result.dossier).toEqual(result.assembly);
|
||||
});
|
||||
|
||||
it("reports missing ownership support instead of guessing endpoint semantics", async () => {
|
||||
const { stdout } = await execute(
|
||||
"python3",
|
||||
[probe, bridge, "no_ownership"],
|
||||
{
|
||||
encoding: "utf8",
|
||||
timeout: 3_000,
|
||||
maxBuffer: 1_024 * 1_024,
|
||||
},
|
||||
);
|
||||
expect(
|
||||
z.object({ error: z.string() }).parse(JSON.parse(stdout)).error,
|
||||
).toBe("Instruction enumeration requires Hopper basic-block ownership");
|
||||
});
|
||||
});
|
||||
+191
@@ -0,0 +1,191 @@
|
||||
"""Exercise the real bridge instruction walker with Hopper-shaped objects."""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
import sys
|
||||
|
||||
|
||||
class Block:
|
||||
def __init__(self, key, start, end, members):
|
||||
self.key, self.start, self.end, self.members = key, start, end, members
|
||||
|
||||
def __eq__(self, other):
|
||||
return isinstance(other, Block) and self.key == other.key
|
||||
|
||||
def __hash__(self):
|
||||
return hash(self.key)
|
||||
|
||||
def getStartingAddress(self):
|
||||
return self.start
|
||||
|
||||
def getEndingAddress(self):
|
||||
return self.end
|
||||
|
||||
def getSuccessorCount(self):
|
||||
return 0
|
||||
|
||||
|
||||
class Instruction:
|
||||
def __init__(self, mnemonic, length=4):
|
||||
self.mnemonic, self.length = mnemonic, length
|
||||
|
||||
def getInstructionLength(self):
|
||||
return self.length
|
||||
|
||||
def getInstructionString(self):
|
||||
return self.mnemonic
|
||||
|
||||
def getArgumentCount(self):
|
||||
return 0
|
||||
|
||||
|
||||
class Segment:
|
||||
def __init__(self, instructions):
|
||||
self.instructions = instructions
|
||||
|
||||
def getInstructionAtAddress(self, address):
|
||||
return self.instructions.get(address)
|
||||
|
||||
def getNameAtAddress(self, address):
|
||||
return "fixture"
|
||||
|
||||
def getReferencesFromAddress(self, address):
|
||||
return [0x9000] if self.instructions[address].mnemonic.startswith("b") else []
|
||||
|
||||
def getReferencesOfAddress(self, address):
|
||||
return []
|
||||
|
||||
def getCommentAtAddress(self, address):
|
||||
return None
|
||||
|
||||
def getInlineCommentAtAddress(self, address):
|
||||
return None
|
||||
|
||||
|
||||
class Procedure:
|
||||
def __init__(self, blocks, segment):
|
||||
self.blocks, self.segment = blocks, segment
|
||||
|
||||
def basicBlockIterator(self):
|
||||
return iter(self.blocks)
|
||||
|
||||
def getBasicBlockAtAddress(self, address):
|
||||
for block in self.blocks:
|
||||
if address in block.members:
|
||||
# Hopper returns fresh wrappers whose native handles compare equal.
|
||||
return Block(block.key, block.start, block.end, block.members)
|
||||
return None
|
||||
|
||||
def getBasicBlockCount(self):
|
||||
return len(self.blocks)
|
||||
|
||||
def getAllCallees(self):
|
||||
return []
|
||||
|
||||
def getAllCallers(self):
|
||||
return []
|
||||
|
||||
def getSegment(self):
|
||||
return self.segment
|
||||
|
||||
def getEntryPoint(self):
|
||||
return self.blocks[0].start
|
||||
|
||||
def signatureString(self):
|
||||
return "int fixture()"
|
||||
|
||||
def getLocalVariableList(self):
|
||||
return []
|
||||
|
||||
def getAllCallerProcedures(self):
|
||||
return []
|
||||
|
||||
def getAllCalleeProcedures(self):
|
||||
return []
|
||||
|
||||
def decompile(self):
|
||||
return "fixture pseudocode"
|
||||
|
||||
|
||||
def main():
|
||||
path = sys.argv[1]
|
||||
bridge = {"__file__": path, "__name__": "rea_hopper_bridge"}
|
||||
exec(compile(Path(path).read_text(encoding="utf-8"), path, "exec"), bridge)
|
||||
scenario = sys.argv[2]
|
||||
instructions = {0x1000: Instruction("mov"), 0x1004: Instruction("b.ne"),
|
||||
0x1008: Instruction("ret"), 0x100C: Instruction("nop")}
|
||||
if scenario == "inclusive":
|
||||
blocks = [Block(0, 0x1000, 0x1004, {0x1000, 0x1004}),
|
||||
Block(1, 0x1008, 0x1008, {0x1008})]
|
||||
elif scenario == "exclusive":
|
||||
blocks = [Block(0, 0x1000, 0x1008, {0x1000, 0x1004}),
|
||||
Block(1, 0x1008, 0x100C, {0x1008})]
|
||||
elif scenario == "duplicate":
|
||||
block = Block(0, 0x1000, 0x1008, {0x1000, 0x1004, 0x1008})
|
||||
blocks = [block, block]
|
||||
elif scenario == "gap":
|
||||
instructions.pop(0x1004)
|
||||
blocks = [Block(0, 0x1000, 0x1008, {0x1000, 0x1004, 0x1008})]
|
||||
elif scenario == "zero_length":
|
||||
instructions[0x1004] = Instruction("invalid", 0)
|
||||
blocks = [Block(0, 0x1000, 0x1008, {0x1000, 0x1004, 0x1008})]
|
||||
elif scenario == "negative_length":
|
||||
instructions[0x1004] = Instruction("invalid", -4)
|
||||
blocks = [Block(0, 0x1000, 0x1008, {0x1000, 0x1004, 0x1008})]
|
||||
elif scenario == "cross_boundary":
|
||||
instructions[0x1000] = Instruction("invalid", 8)
|
||||
blocks = [Block(0, 0x1000, 0x1004, {0x1000})]
|
||||
elif scenario == "foreign_owner":
|
||||
blocks = [Block(0, 0x1000, 0x100C, {0x1000, 0x1004})]
|
||||
elif scenario == "reversed":
|
||||
blocks = [Block(0, 0x1008, 0x1000, {0x1008})]
|
||||
elif scenario == "variable_length":
|
||||
instructions = {0x1000: Instruction("mov", 3),
|
||||
0x1003: Instruction("b.ne", 2),
|
||||
0x1005: Instruction("ret", 1),
|
||||
0x1006: Instruction("nop", 1)}
|
||||
blocks = [Block(0, 0x1000, 0x1003, {0x1000, 0x1003}),
|
||||
Block(1, 0x1005, 0x1005, {0x1005})]
|
||||
elif scenario == "no_ownership":
|
||||
blocks = [Block(0, 0x1000, 0x1004, {0x1000})]
|
||||
else:
|
||||
raise ValueError("Unknown scenario")
|
||||
segment = Segment(instructions)
|
||||
procedure = Procedure(blocks, segment)
|
||||
if scenario == "no_ownership":
|
||||
procedure.getBasicBlockAtAddress = None
|
||||
try:
|
||||
bridge["_instruction_addresses"](procedure)
|
||||
except bridge["CapabilityUnavailableError"] as error:
|
||||
print(json.dumps({"error": str(error)}))
|
||||
return
|
||||
raise AssertionError("Missing ownership API was silently accepted")
|
||||
class API:
|
||||
def require_analysis_complete(self, document, method):
|
||||
pass
|
||||
|
||||
class Document:
|
||||
def getSegmentsList(self):
|
||||
return []
|
||||
|
||||
document = Document()
|
||||
bridge["_api"] = lambda: API()
|
||||
bridge["_document"] = lambda name=None: document
|
||||
bridge["_procedure"] = lambda document, value=None: procedure
|
||||
bridge["_segment"] = lambda document, address: segment
|
||||
bridge["_containing_procedure"] = lambda document, address: (None, None)
|
||||
bridge["_search_inventory"] = lambda document, kind: []
|
||||
dossier = bridge["_analyze_function"](document, {})
|
||||
print(json.dumps({
|
||||
"blocks": dossier["basic_blocks"],
|
||||
"length": bridge["_dispatch"]("procedure_info", {})["length"],
|
||||
"addresses": bridge["_instruction_addresses"](procedure),
|
||||
"assembly": bridge["_assembly"](procedure).splitlines(),
|
||||
"fast": bridge["_read_function_instructions"](document, {})["instructions"],
|
||||
"dossier": dossier["assembly"],
|
||||
"references": bridge["_procedure_references"](document, {})["references"],
|
||||
}))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user