Edward J. Schwartz
Show memory writers and highlight corresponding candidate C lines
c5747c7
Raw History Blame Contribute Delete
29.3 kB
"""Local Gradio demo. Run inside the Linux x86-64 Docker environment."""
import argparse
import json
from contextlib import closing
from dataclasses import asdict
from hashlib import sha256
from decimal import Decimal
import re
import shutil
from pathlib import Path
import verifier
import shlex
import tempfile
import gradio as gr
from demo import examples as example_catalog
from demo.examples import EXAMPLE_LABELS, ROOT, prepare_examples, reference_sources, example_groups
from demo.presentation import banner, format_bit_ranges, json_detail, mismatch_summary, status
from demo.session import (UISession, begin_ui, cancel_session, controls, edit_inputs, idle_controls,
reset_session, stream_ui)
import verifier as api
from demo.execution import verification_reports
from verifier import (Task, ExecutionConfig, SamplingConfig, GuidanceConfig, ConcolicConfig, InputRecipe,
integer_text, parse_bindings, parse_function, yaml_mapping)
HEADERS = ["Execution", "Source", "Verdict", "Location", "Differing bits", "Original", "Candidate", "Reason"]
def integer(value, label):
if type(value) is int:
return value
if isinstance(value, float) and value.is_integer():
return int(value)
if isinstance(value, str):
try:
if value.strip().lower().startswith("0x"):
return int(value.strip(), 16)
if re.fullmatch(r"[+-]?[0-9]+", value.strip()):
return int(Decimal(value.strip()))
except ValueError:
pass
raise ValueError(f"{label} must be an integer")
def execution_source(report):
if report.task.get("scheduler") == "fixed_input_replay":
return "Replay"
testcases = getattr(report, "testcases", ())
origin = testcases[-1].origin if testcases else "random"
if origin == "generated":
return "Concolic"
if origin == "replay":
return "Replay"
if report.task.get("execution", {}).get("mode") == "comparison_guided":
return "Comparison guided"
return "Random"
def result_row(result, execution_index=None, source="Random"):
diff = result.difference
bits = format_bit_ranges((span.first, span.last) for span in getattr(diff, "bit_ranges", ()))
original, candidate = getattr(diff, "original", ""), getattr(diff, "candidate", "")
if isinstance(original, int):
original, candidate = hex(original), hex(candidate)
return [integer_text(result.seed if execution_index is None else execution_index), source, result.verdict, getattr(diff, "location", ""), bits,
original, candidate, result.reason or result.original.reason or result.candidate.reason or ""]
def task_from_inputs(binary, function, candidate, bindings, candidate_symbol,
execution_count, seed_start, instruction_budget, compiler, flags, interfaces="",
execution_mode="comparison_guided", null_probability=0,
repeated_branch_threshold=8, max_learned_probability=0.75,
replay_recipe=None, max_new_testcases_per_run=8, solver_timeout_ms=1000,
random_input_probability=0.25):
if not binary:
raise ValueError("upload a binary or select an example")
options = yaml_mapping(interfaces or "")
allowed = {"original_prototype", "original_cc", "candidate_cc", "variadic_types", "callee_interfaces", "memory_ranges", "original_base"}
if set(options) - allowed:
raise ValueError("unknown interface settings: " + ", ".join(sorted(set(options) - allowed)))
sampling = (SamplingConfig(null_probability, null_probability, null_probability)
if execution_mode == "random" else SamplingConfig())
guidance = (GuidanceConfig(integer(repeated_branch_threshold, "repetition threshold"), max_learned_probability)
if execution_mode == "comparison_guided" else GuidanceConfig())
concolic = (ConcolicConfig(integer(max_new_testcases_per_run, "max new testcases per run"),
integer(solver_timeout_ms, "solver timeout"), random_input_probability)
if execution_mode == "concolic" else ConcolicConfig())
return Task(binary=Path(binary), function=parse_function(function), candidate=candidate,
bindings=parse_bindings(bindings), candidate_symbol=candidate_symbol.strip(),
execution_count=integer(execution_count, "execution count"), seed_start=integer(seed_start, "seed start"),
instruction_budget=integer(instruction_budget, "instruction budget"),
compiler=compiler, flags=tuple(shlex.split(flags)),
execution=ExecutionConfig(execution_mode, sampling, guidance, concolic),
**options)
def load_recipe(path):
data = json.loads(Path(path).read_text(), parse_int=lambda token: int(Decimal(token)))
if isinstance(data, dict) and "recipes" in data:
recipes = data["recipes"]
if not isinstance(recipes, list) or not recipes:
raise ValueError("the report has no saved input recipes")
data = recipes[0]
return InputRecipe.from_mapping(data)
def clear_outputs():
return "", [], None, None, "", ""
def show_assembly(binary, function, candidate, bindings, candidate_symbol,
execution_count, seed_start, instruction_budget, compiler, flags, interfaces="",
execution_mode="comparison_guided", null_probability=0,
repeated_branch_threshold=8, max_learned_probability=0.75,
replay_recipe=None, max_new_testcases_per_run=8, solver_timeout_ms=1000,
random_input_probability=0.25):
try:
report = api.inspect(task_from_inputs(binary, function, candidate, bindings, candidate_symbol,
execution_count, seed_start, instruction_budget, compiler, flags, interfaces,
execution_mode, null_probability, repeated_branch_threshold,
max_learned_probability, replay_recipe, max_new_testcases_per_run,
solver_timeout_ms, random_input_probability))
yield banner(report, preview=True), [], None, None, report.original_assembly, report.candidate_assembly
except (ValueError, OSError) as error:
raise gr.Error(str(error)) from error
def run_verification(binary, function, candidate, bindings, candidate_symbol,
execution_count, seed_start, instruction_budget, compiler, flags, interfaces="",
execution_mode="comparison_guided", null_probability=0,
repeated_branch_threshold=8, max_learned_probability=0.75,
replay_recipe=None, max_new_testcases_per_run=8, solver_timeout_ms=1000,
random_input_probability=0.25, *, report_directory=None, cancelled=None):
"""Translate UI inputs, stream engine results, then supply the JSON download."""
try:
task = task_from_inputs(binary, function, candidate, bindings, candidate_symbol,
execution_count, seed_start, instruction_budget, compiler, flags, interfaces,
execution_mode, null_probability, repeated_branch_threshold, max_learned_probability, replay_recipe,
max_new_testcases_per_run, solver_timeout_ms, random_input_probability)
recipe = load_recipe(replay_recipe) if replay_recipe else None
rows, detail, report = [], None, None
with closing(verification_reports(task, cancelled=cancelled, recipe=recipe)) as iterator:
for report in iterator:
rows.append(result_row(report.results[-1], len(rows) + 1, execution_source(report)))
if detail is None and report.first_mismatch is not None:
detail = asdict(report.first_mismatch)
yield banner(report), list(rows), detail, None, report.original_assembly, report.candidate_assembly
if cancelled is not None and cancelled.is_set():
return
output = report.to_json()
directory = Path(report_directory) if report_directory else Path(tempfile.gettempdir()) / "decompile-verify-reports"
directory.mkdir(parents=True, exist_ok=True)
path = directory / (sha256(output.encode()).hexdigest() + ".json")
# Atomic writes allow concurrent requests for the same deterministic report.
with tempfile.NamedTemporaryFile(mode="w", dir=directory, delete=False) as stream:
stream.write(output)
temporary = Path(stream.name)
temporary.replace(path)
yield banner(report, complete=True), rows, detail, str(path), report.original_assembly, report.candidate_assembly
except (ValueError, OSError) as error:
raise gr.Error(str(error)) from error
class Workbench(gr.Blocks):
"""Apply the workbench theme even when a caller launches create_app() directly."""
def launch(self, **kwargs):
kwargs.setdefault("theme", gr.themes.Base(
primary_hue="teal", secondary_hue="teal", neutral_hue="slate",
font=["system-ui", "sans-serif"], font_mono=["ui-monospace", "monospace"],
))
kwargs.setdefault("css_paths", Path(__file__).with_name("workbench.css"))
kwargs.setdefault("js", Path(__file__).with_name("workbench.js").read_text())
blocked = list(kwargs.pop("blocked_paths", ()) or ())
blocked.extend([str(Path(verifier.__file__).resolve().parent),
str(ROOT / "core"), str(ROOT / ".git")])
helper = shutil.which("decompile-verify-declarations")
if helper:
blocked.append(str(Path(helper).resolve()))
kwargs["blocked_paths"] = blocked
return super().launch(**kwargs)
def create_app(examples_directory=None):
examples = prepare_examples(examples_directory or ROOT / "build" / "demo-examples")
references = reference_sources()
with Workbench(title="Decompiler verifier", analytics_enabled=False,
fill_width=True, delete_cache=(3600, 86400)) as app:
session = gr.State(UISession())
ticket = gr.State(None)
reference_function = gr.State(None)
gr.Markdown("""# Decompiler verifier
This tool takes an **executable file**, a **function in that executable**, and a **candidate C
implementation**, and attempts to determine whether the candidate and the original function
are equivalent. It compiles the candidate C, runs both functions on the same seeded random
inputs, and compares their behavior. A mismatch shows where they differ; agreement means
they matched on the tested inputs within the comparison scope described below.
**Quick start:** Pick an **example** below and click **Verify**.
""", elem_id="intro")
with gr.Accordion("About bindings, run settings, and your own inputs", open=False):
gr.Markdown("""Open **Bindings** to map candidate `extern` names to original binary addresses: `globals`
for shared data and `callees` for called functions. Bound callees receive deterministic
synthetic returns. In the same panel, `original_prototype` is the **trusted prototype**: a C
declaration describing the original function's argument and return types, such as
`unsigned original(unsigned x);`. Supply it from your knowledge of the binary, independently
of the candidate's guessed types; it controls input generation and return interpretation.
The examples supply it. You can also specify callee interfaces and memory ranges there.
Open **Run settings** to adjust seeds, instruction budget, compiler, or flags, then click
**Verify**. Inspect the assembly panes, per-seed results, and first mismatch, or download the
JSON report. Agreement covers the tested inputs and comparison scope described below.
**Try changing the Candidate C**—an operator, constant, or type—and verify again with the
same seeds to see how the assembly and results change.
For your own input, upload an **x86-64 ELF binary**, enter the **function start address**,
and paste complete **Candidate C**. Set **Candidate function name** to the C function to test.
The examples include matching implementations, deliberate bugs, real `factor` binary
functions, paper and artifact cases under **Related work**, and saved **Ghidra exports** in expandable sections.
Some Ghidra exports agree; others reveal mismatches or compile/type errors. **Reference C**, when shown, is a reading
aid; verification runs the binary.
""")
# Render inputs later so the example picker can sit above setup.
binary = gr.File(label="Binary", type="filepath", file_count="single", render=False, elem_id="binary")
function = gr.Textbox(label="Function start address", placeholder="0x401000",
info="angr recovers the function's blocks using available symbols and control flow.", render=False)
candidate_symbol = gr.Textbox(label="Candidate function name", value="target", render=False)
candidate = gr.Code(label="Candidate C", language="c", lines=20, wrap_lines=False,
buttons=["copy"], render=False, elem_id="candidate-source")
bindings = gr.Code(label="Bindings (YAML)", language="yaml", lines=4,
value="globals: {}\ncallees: {}\n", render=False)
execution_count = gr.Number(label="Number of executions", value=100, precision=0, minimum=1, render=False)
seed_start = gr.Textbox(label="Initial seed", value="0", info="Nonnegative decimal or hex integer", render=False)
instruction_budget = gr.Number(label="Instruction budget per run", value=1000, precision=0,
minimum=1, render=False)
compiler = gr.Dropdown(label="Candidate compiler", choices=["gcc", "clang"], value="gcc", render=False)
flags = gr.Textbox(label="Candidate flags", value="-O2",
info="Optimization/debug flags and selected C options; see docs/demo.md.", render=False)
interfaces = gr.Code(label="Function interfaces and explicit memory ranges (YAML)", language="yaml",
value="", lines=5, render=False)
example_inputs = [binary, function, candidate, bindings, candidate_symbol,
execution_count, seed_start, instruction_budget, compiler, flags, interfaces]
execution_mode = gr.Dropdown(label="Execution mode", choices=["random", "comparison_guided", "concolic"],
value="comparison_guided", render=False,
info="random samples inputs with a configurable NULL probability. "
"comparison_guided learns preferred values from branch comparisons. "
"concolic solves branch conditions to generate testcases.")
null_probability = gr.Slider(label="NULL probability", minimum=0, maximum=1, value=0,
step=0.01, visible=False, render=False,
info="Chance of choosing NULL (zero) for each new argument value, lazy memory word, "
"or call response. Otherwise, sample randomly. Uses zero for non-pointer values.")
repeated_branch_threshold = gr.Number(label="Repeated branch threshold", value=8,
precision=0, minimum=1, render=False,
info="Consecutive identical outcomes at the same branch before learning "
"preferred values. Lower values start learning sooner.")
max_learned_probability = gr.Slider(label="Learned-value probability", minimum=0, maximum=1,
value=0.75, step=0.01, render=False,
info="Once relevant values are learned, chance of choosing one of them. "
"Otherwise, sample randomly.")
replay_recipe = gr.File(label="Replay recipe or report (JSON)", type="filepath",
file_types=[".json"], render=False)
max_new_testcases_per_run = gr.Number(label="Max new testcases per run", value=8, precision=0,
minimum=1, render=False)
solver_timeout_ms = gr.Number(label="Solver timeout per query (ms)", value=1000, precision=0,
minimum=1, render=False)
random_input_probability = gr.Slider(label="Fresh random input probability", minimum=0, maximum=1,
value=0.25, step=0.01, render=False)
mode_inputs = [execution_mode, null_probability,
repeated_branch_threshold, max_learned_probability, replay_recipe,
max_new_testcases_per_run, solver_timeout_ms, random_input_probability]
inputs = [*example_inputs, *mode_inputs]
gr.Markdown("### Examples\nOpen a section and select an example.", elem_id="examples")
gr.HTML('<div class="example-legend" aria-label="Expected example results">'
'<span>Expected with bundled settings:</span>'
'<span class="indicator green">Agree</span>'
'<span class="indicator amber">Inconclusive / error</span>'
'<span class="indicator red">Mismatch</span></div>',
apply_default_css=False)
example_pickers = []
for group_number, (label, indices) in enumerate(example_groups(example_catalog.EXAMPLE_NAMES)):
with gr.Accordion(f"{label} ({len(indices)})", open=label == "Basics"):
gr.Markdown(example_catalog.EXAMPLE_GROUP_DESCRIPTIONS[label],
elem_classes=["example-description"])
expected = [example_catalog.expected_result(example_catalog.EXAMPLE_NAMES[index],
examples[index][8]) for index in indices]
picker = gr.Examples(examples=[examples[index] for index in indices], inputs=example_inputs,
label=label, example_labels=[
f"{result.capitalize()} · {EXAMPLE_LABELS[index]}"
for index, result in zip(indices, expected)],
examples_per_page=len(indices), cache_examples=False,
elem_id=f"examples-group-{group_number}")
example_pickers.append((picker, indices))
with gr.Row(elem_id="setup"):
binary.render()
function.render()
candidate_symbol.render()
with gr.Accordion("Bindings", open=False):
bindings.render()
interfaces.render()
with gr.Accordion("Run settings", open=False):
with gr.Row():
execution_count.render()
seed_start.render()
instruction_budget.render()
compiler.render()
flags.render()
execution_mode.render()
null_probability.render()
with gr.Column(visible=True) as guidance_settings:
with gr.Row():
repeated_branch_threshold.render()
max_learned_probability.render()
gr.Markdown("comparison_guided starts with random inputs. After repeated branch outcomes, "
"it favors values from comparisons (such as NULL) for future inputs. "
"Already assigned values stay fixed.")
with gr.Column(visible=False) as concolic_settings:
with gr.Row():
max_new_testcases_per_run.render()
solver_timeout_ms.render()
random_input_probability.render()
gr.Markdown("Each execution tests one input against both functions and collects branch constraints. "
"Choose a fresh random input with this probability; otherwise try a queued testcase. "
"An empty queue always starts with a random input. Agreement covers tested inputs.")
execution_mode.change(lambda mode: (gr.update(visible=mode == "random"),
gr.update(visible=mode == "comparison_guided"),
gr.update(visible=mode == "concolic")),
execution_mode, [null_probability, guidance_settings, concolic_settings], queue=False,
api_visibility="private", show_progress="hidden")
gr.Markdown("Uploading a recipe skips exploration and replays one input; "
"uploading a report replays its first recipe.")
replay_recipe.render()
with gr.Row(elem_id="actions"):
run = gr.Button("Verify", variant="primary")
cancel = gr.Button("Cancel", interactive=False)
verdict = gr.HTML(status("Ready · Choose an example or upload a binary"), elem_id="verdict",
apply_default_css=False)
with gr.Row(elem_id="workspace") as workspace:
candidate.render()
reference = gr.Code(label="Reference C", language="c", lines=20,
interactive=False, wrap_lines=False, buttons=["copy"],
visible=False, elem_id="reference-source")
original_assembly = gr.Code(label="Original assembly", language=None, lines=20,
interactive=False, wrap_lines=False, buttons=["copy"],
show_line_numbers=False, elem_id="original-assembly")
candidate_assembly = gr.Code(label="Candidate assembly", language=None, lines=20,
interactive=False, wrap_lines=False, buttons=["copy"],
show_line_numbers=False, elem_id="candidate-assembly")
gr.Markdown("Compares ordered calls, non-frame memory at calls and return, typed returns, "
"and termination. Bound callees use declared signatures and deterministic "
"synthetic returns without memory effects. Frame-observing and unsupported "
"calls are inconclusive.", elem_id="scope")
with gr.Row(elem_id="results"):
table = gr.Dataframe(headers=HEADERS, datatype=["str"] * len(HEADERS),
interactive=False, label="Execution results", scale=2,
min_width=320, elem_id="seed-table", max_height=340,
column_widths=[90, 160, 100, 100, 140, 120, 120, 240])
with gr.Column(scale=1, min_width=280):
gr.Markdown("### First mismatch")
summary = gr.HTML(mismatch_summary(None), apply_default_css=False, elem_id="mismatch-summary")
with gr.Accordion("Full mismatch JSON", open=False):
detail = gr.JSON(label="First mismatch")
download = gr.DownloadButton(label="Download JSON report")
outputs = [verdict, table, detail, download, original_assembly, candidate_assembly]
buttons = [run, cancel]
def verify_ui(binary, function, candidate, bindings, candidate_symbol,
execution_count, seed_start, instruction_budget, compiler, flags, interfaces="",
execution_mode="comparison_guided", null_probability=0,
repeated_branch_threshold=8, max_learned_probability=0.75,
replay_recipe=None, max_new_testcases_per_run=8, solver_timeout_ms=1000,
random_input_probability=0.25, session=None, ticket=None):
with closing(stream_ui(run_verification, (binary, function, candidate, bindings, candidate_symbol,
execution_count, seed_start, instruction_budget, compiler, flags, interfaces,
execution_mode, null_probability, repeated_branch_threshold,
max_learned_probability, replay_recipe, max_new_testcases_per_run,
solver_timeout_ms, random_input_probability),
session, ticket, cancellable=True)) as updates:
for update in updates:
yield update[0], update[1], json_detail(update[2]), *update[3:]
def reset_ui(session):
reset_session(session)
return clear_outputs()
def reference_loader(indices):
def load_reference(example):
index, values = example
source = references[indices[index]]
return (gr.update(value=source, visible=bool(source)),
gr.update(elem_classes=["has-reference"] if source else []), values[1])
return load_reference
def clear_reference():
return gr.update(value="", visible=False), gr.update(elem_classes=[]), None
def clear_changed_reference(function, reference_function):
if reference_function is not None and function != reference_function:
return clear_reference()
return (gr.skip(),) * 3
reference_outputs = [reference, workspace, reference_function]
# State inputs are private to Gradio and omitted from the public API signature.
verify_start = run.click(begin_ui, [session, *inputs], [verdict, *buttons, ticket],
queue=False, api_visibility="private", show_progress="hidden")
preview_event = gr.on(triggers=[], fn=show_assembly, inputs=inputs, outputs=outputs,
api_name="assembly", concurrency_limit=None, show_progress="hidden")
event = verify_start.then(verify_ui, [*inputs, session, ticket], outputs, api_name="verify",
concurrency_limit=None, show_progress="hidden")
event.then(idle_controls, session, buttons, queue=False,
api_visibility="private", show_progress="hidden")
gr.on([binary.upload, binary.delete, binary.clear,
replay_recipe.upload, replay_recipe.delete, replay_recipe.clear,
*[component.input for component in inputs[1:] if component is not replay_recipe]],
edit_inputs, [session, *inputs], verdict,
queue=False, api_visibility="private", show_progress="hidden")
def render_mismatch(detail, session):
with session.lock:
source = session.snapshot[2] if session.snapshot is not None else None
return mismatch_summary(detail, source)
detail.change(render_mismatch, [detail, session], summary, queue=False,
api_visibility="private", show_progress="hidden")
example_clicks = [picker.dataset.click for picker, _ in example_pickers]
gr.on(example_clicks, reset_ui, session, outputs, queue=False,
cancels=[event, preview_event], api_name="reset_outputs",
api_visibility="undocumented", show_progress="hidden")
gr.on(example_clicks, lambda: controls(False), outputs=buttons, queue=False,
api_visibility="private", show_progress="hidden")
for picker, indices in example_pickers:
picker.dataset.click(reference_loader(indices), picker.dataset, reference_outputs,
queue=False, api_visibility="private", show_progress="hidden")
gr.on([binary.upload, binary.delete, binary.clear, function.input],
clear_reference, outputs=reference_outputs, queue=False,
api_visibility="private", show_progress="hidden")
# Gradio 6.9 Textbox.input fires on keypress; blur also catches paste/deletion.
function.blur(clear_changed_reference, [function, reference_function], reference_outputs,
queue=False, api_visibility="private", show_progress="hidden")
cancel.click(cancel_session, session, verdict, cancels=[event, preview_event], queue=False,
api_visibility="private", show_progress="hidden").then(
idle_controls, session, buttons, queue=False,
api_visibility="private", show_progress="hidden")
return app.queue(max_size=16)
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=7860)
args = parser.parse_args()
create_app().launch(server_name=args.host, server_port=args.port, share=False)
if __name__ == "__main__":
main()