show_function: the source is the unit the DWARF names for the function's address

The source body and the source graph each took the first preprocessed unit, in discovery
order, whose text or CFGs held the function's name, so a static function defined in
several units showed another unit's body and graph. The unit is now the owner the binary's
DWARF gives the address (binfmt.source_function_owners, with the run's abstract-origin
policy), as ged_causes and counterfactual_ged already take it; a function with no owner
prints "no DWARF owner" and no source. The DWARF variables came from a loop that accepted
any address whose only function carried the name; they are now what the type metric's own
_ground_truth_for_function joins for the address, so the display shows what was scored.

Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01456qmqCGpCWJkrg6JUZNNM

tests/decbench/show_function.py

@@ -30,7 +30,9 @@ def main(argv: list[str] | None = None) -> int:
     from decbench.models.project import OptimizationLevel, Project
     from decbench.pipeline.executor import PipelineConfig, PipelineExecutor
     from decbench.results_store import gather_project_tomls
-    from decbench.metrics.type_match import extract_ground_truth_type_index
+    from decbench.metrics.type_match import _ground_truth_for_function, extract_ground_truth_type_index
+    from decbench.utils.binfmt import detect, source_function_owners
+    from decbench.utils.dwarf_policy import dwarf_follow_abstract_origin
 
     ck = pickle.loads((a.run / "checkpoints" / f"{a.project}.pkl").read_bytes())
     opt = OptimizationLevel(a.opt)
@@ -43,33 +45,34 @@ def main(argv: list[str] | None = None) -> int:
         if m in ev and a.function in ev[m].function_results:
             v = ev[m].function_results[a.function]
             print(f"{m}: {v.value}  {({k: v.metadata[k] for k in v.metadata if k in ('tp','fp','fn','gt_vars','source_nodes','source_edges','decompiled_nodes','decompiled_edges')})}")
-    gt = extract_ground_truth_type_index(res.binary_path)
-    for addr, names in gt.items():
-        if a.function in names and (addr == fn.address or len(names) == 1):
-            print("DWARF variables:")
-            for v in names[a.function]:
-                print("  ", v["name"], "arg" if v["is_arg"] else "loc", v["type"][:5], "stack" if v["rbp_offset"] else "")
+    info = detect(res.binary_path)
+    variables = _ground_truth_for_function(extract_ground_truth_type_index(res.binary_path), a.function,
+                                           int(fn.address), info.arch if info is not None else "")
+    if variables:
+        print("DWARF variables:")
+        for v in variables:
+            print("  ", v["name"], "arg" if v["is_arg"] else "loc", v["type"][:5], "stack" if v["rbp_offset"] else "")
+    # The unit that defines the function is the one the binary's DWARF names for its address, never a guess by name.
+    units, owner = {}, None
     if not a.no_source or a.graphs:
         project = Project.from_toml(next(t for t in gather_project_tomls() if t.stem == a.project))
         PipelineExecutor(PipelineConfig(output_dir=a.run, optimization_levels=[opt]))._discover_existing_binaries([project], a.run)
-    if not a.no_source:
-        for stem, path in project.preprocessed_sources.get(opt, {}).items():
-            body = function_body(Path(path).read_text(errors="replace"), a.function)
-            if body:
-                print(f"--- source ({stem}.i)")
-                print(body)
-                break
+        units = {stem: Path(path) for stem, path in project.preprocessed_sources.get(opt, {}).items()}
+        owner = source_function_owners(
+            res.binary_path, set(units), follow_abstract_origin=dwarf_follow_abstract_origin()).get(int(fn.address))
+        if owner is None:
+            print(f"--- source: no DWARF owner at {hex(fn.address)}")
+    if owner is not None and not a.no_source:
+        body = function_body(units[owner[1]].read_text(errors="replace"), owner[0])
+        if body:
+            print(f"--- source ({units[owner[1]].name})")
+            print(body)
     if a.graphs:
         from decbench.utils.cfg import extract_cfgs_from_decompilation, extract_cfgs_from_source
 
         sub = res.model_copy(update={"functions": {a.function: fn}})
         dgraph = extract_cfgs_from_decompilation(sub).get(a.function)
-        sgraph = None
-        for stem, path in project.preprocessed_sources.get(opt, {}).items():
-            sg = extract_cfgs_from_source(Path(path))
-            if a.function in sg:
-                sgraph = sg[a.function]
-                break
+        sgraph = extract_cfgs_from_source(units[owner[1]]).get(owner[0]) if owner is not None else None
         for label, g in (("source", sgraph), (a.dec, dgraph)):
             if g is None:
                 continue

tests/decbench/test_show_function.py

@@ -74,8 +74,14 @@ class Shown(unittest.TestCase):
         (self.run / "checkpoints").mkdir()
         (self.run / "checkpoints" / "zlib.pkl").write_bytes(b"")
         self.loaded = []
-        self.source_graphs = {"crc32": chain(3)}
-        self.units = {"crc32": Path("crc32.i")}
+        self.owner_calls, self.variable_calls = [], []
+        # Two units define a crc32, `alpha` sorts first; the DWARF says the function at 0x1000 is zeta's.
+        self.units = {}
+        for stem, value in (("alpha", 1), ("zeta", 2)):
+            (self.run / f"{stem}.i").write_text(f"int crc32(void) {{ return {value}; }}\n")
+            self.units[stem] = self.run / f"{stem}.i"
+        self.source_graphs = {"alpha": {"crc32": chain(7)}, "zeta": {"crc32": chain(3)}}
+        self.owners = {0x1000: ("crc32", "zeta")}
 
         class Project:
             preprocessed_sources = {"O2": self.units}
@@ -93,6 +99,14 @@ class Shown(unittest.TestCase):
                 pass
 
         graphs = {"crc32": chain(2)}
+
+        def source_function_owners(path, stems, follow_abstract_origin=False):
+            self.owner_calls.append((path, set(stems), follow_abstract_origin))
+            return self.owners
+
+        def ground_truth_for_function(index, name, address, architecture):
+            self.variable_calls.append((name, address, architecture))
+            return [{"name": "strm", "is_arg": True, "type": ["z_stream*"], "rbp_offset": 0}]
         stand_ins = {
             "decbench": module("decbench"), "decbench.models": module("decbench.models"),
             "decbench.pipeline": module("decbench.pipeline"), "decbench.metrics": module("decbench.metrics"),
@@ -103,10 +117,16 @@ class Shown(unittest.TestCase):
             "decbench.results_store": module(
                 "decbench.results_store", gather_project_tomls=lambda: [Path("tomls/zlib.toml")]),
             "decbench.metrics.type_match": module(
-                "decbench.metrics.type_match", extract_ground_truth_type_index=lambda path: {}),
+                "decbench.metrics.type_match", extract_ground_truth_type_index=lambda path: {},
+                _ground_truth_for_function=ground_truth_for_function),
             "decbench.utils.cfg": module(
                 "decbench.utils.cfg", extract_cfgs_from_decompilation=lambda result: graphs,
-                extract_cfgs_from_source=lambda path: self.source_graphs),
+                extract_cfgs_from_source=lambda path: self.source_graphs[Path(path).stem]),
+            "decbench.utils.binfmt": module(
+                "decbench.utils.binfmt", source_function_owners=source_function_owners,
+                detect=lambda path: types.SimpleNamespace(arch="x86")),
+            "decbench.utils.dwarf_policy": module(
+                "decbench.utils.dwarf_policy", dwarf_follow_abstract_origin=lambda: True),
         }
         checkpoint = {"decompile": {"O2": {"libz.so.1.2": {"r2sleigh_native": Result()}}},
                       "evaluate": {"O2": {"libz.so.1.2": {"r2sleigh_native": {}}}}}
@@ -133,6 +153,26 @@ class Shown(unittest.TestCase):
         self.assertIn("--- r2sleigh_native block graph: 2 blocks, 1 edges", text)
         self.assertNotIn("--- source (", text)
 
+    def test_the_source_is_the_unit_the_dwarf_names_for_the_address_not_the_first_that_has_the_name(self):
+        text = self.show("--graphs")
+        self.assertIn("--- source (zeta.i)\n{ return 2; }", text)
+        self.assertNotIn("return 1;", text)
+        self.assertIn("--- source block graph: 3 blocks, 2 edges", text)  # zeta's chain(3), not alpha's chain(7)
+        self.assertEqual(self.owner_calls, [(Path("/unused/libz.so.1.2.13"), {"alpha", "zeta"}, True)])
+
+    def test_a_function_whose_address_has_no_owner_shows_no_source_and_says_so(self):
+        self.owners = {}
+        text = self.show("--graphs")
+        self.assertIn("--- source: no DWARF owner at 0x1000", text)
+        self.assertNotIn("--- source (", text)
+        self.assertNotIn("--- source block graph", text)
+        self.assertIn("--- r2sleigh_native block graph: 2 blocks, 1 edges", text)
+
+    def test_the_dwarf_variables_are_the_ones_the_type_metric_joins_by_address(self):
+        text = self.show("--no-source")
+        self.assertEqual(self.variable_calls, [("crc32", 0x1000, "x86")])
+        self.assertIn("DWARF variables:\n   strm arg ['z_stream*'] ", text)
+
     def test_the_project_is_not_loaded_when_neither_source_nor_graphs_are_asked_for(self):
         text = self.show("--no-source")
         self.assertEqual(self.loaded, [])