"""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('
' 'Expected with bundled settings:' 'Agree' 'Inconclusive / error' 'Mismatch
', 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()