Improving ghidra decompilation support (#3269)

* decompiling kernel funcs

* accounting for KASLR

* handling kernel modules

* cleaning up

* doc

* changes based on comments

docs/configuration/config.md

@@ -399,6 +399,18 @@ Display internal event debugging info.
 
 ----------
 
+## **decompiler**
+
+
+Framework that your ghidra plugin installed.
+
+
+
+**Default:** 'radare2'  
+**Valid values:** 'radare2', 'rizin'
+
+----------
+
 ## **default-visualize-chunk-number**
 
 
@@ -828,18 +840,6 @@ Whether to show call arguments below instruction.
 
 ----------
 
-## **r2decompiler**
-
-
-Framework that your ghidra plugin installed.
-
-
-
-**Default:** 'radare2'  
-**Valid values:** 'radare2', 'rizin'
-
-----------
-
 ## **safe-linking**
 
 

pwndbg/aglib/elf.py

@@ -202,6 +202,14 @@ def get_containing_sections(elf_filepath: str, elf_loadaddr: int, vaddr: int):
     return sections
 
 
+def get_vmlinux_unrand_base(elf_filepath: str):
+    elf = get_elf_info(elf_filepath)
+    for seg in elf.segments:
+        if seg["p_type"] == "PT_LOAD":
+            return seg["p_vaddr"]
+    return None
+
+
 def dump_section_by_name(
     filepath: str, section_name: str, try_local_path: bool = False
 ) -> Tuple[int, int, bytes] | None:

pwndbg/aglib/file.py

@@ -90,7 +90,7 @@ def get_file(path: str, try_local_path: bool = False) -> str:
         path = path[7:]  # len('target:') == 7
 
     local_path = path
-    if not pwndbg.aglib.remote.is_remote():
+    if not pwndbg.aglib.remote.is_remote() or pwndbg.aglib.qemu.is_qemu_kernel():
         if not os.path.exists(local_path):
             raise OSError(f"File '{local_path}' does not exist", errno.ENOENT)
 

pwndbg/aglib/kernel/__init__.py

@@ -268,7 +268,7 @@ class ArchOps(ABC):
         return arch_paginginfo().vmemmap
 
     @property
-    def kbase(self) -> int:
+    def kbase(self) -> int | None:
         return arch_paginginfo().kbase
 
     @property
@@ -365,7 +365,7 @@ class x86_64Ops(x86Ops):
         return pwndbg.dbg.selected_inferior().create_value(per_cpu_addr)
 
     def virt_to_phys(self, virt: int) -> int:
-        if virt < self.kbase:
+        if self.kbase is None or virt < self.kbase:
             return (virt - self.page_offset) % (1 << 64)
         return ((virt - self.kbase) + self.phys_base) % (1 << 64)
 

pwndbg/commands/kmod.py

@@ -82,9 +82,13 @@ def kmod(module_name=None, path=None) -> None:
                 table.append([f"{addr:#x}", name, size, "-"])
     if path is not None:
         if len(table) == 1:
-            pwndbg.dbg.selected_inferior().add_symbol_file(path, table[0][0])
-            return
-        if len(table) > 1:
+            addr = table[0][0]
+            pwndbg.dbg.selected_inferior().add_symbol_file(path, addr)
+            if pwndbg.config.decompiler == "radare2":
+                pwndbg.radare2.r2cmd(["o", path, addr])
+            elif pwndbg.config.decompiler == "rizin":
+                pwndbg.rizin.rzcmd(["o", path, addr])
+        elif len(table) > 1:
             print(M.warn("Multiple modules detected with the given filter"))
         else:
             print(M.warn("No modules detected with the given filter."))

pwndbg/commands/radare2.py

@@ -9,7 +9,6 @@ import pwndbg.aglib.proc
 import pwndbg.aglib.regs
 import pwndbg.commands
 import pwndbg.radare2
-from pwndbg.color import message
 from pwndbg.commands import CommandCategory
 
 parser = argparse.ArgumentParser(description="Launches radare2.")
@@ -97,10 +96,4 @@ pwndbg> r2pipe pdf @ sym.main
 )
 @pwndbg.commands.OnlyWithFile
 def r2pipe(arguments) -> None:
-    try:
-        r2 = pwndbg.radare2.r2pipe()
-        print(r2.cmd(" ".join(arguments)))
-    except ImportError:
-        print(message.error("Could not import r2pipe python library. Is it installed?"))
-    except Exception as e:
-        print(message.error(e))
+    print(pwndbg.radare2.r2cmd(arguments))

pwndbg/commands/rizin.py

@@ -9,7 +9,6 @@ import pwndbg.aglib.proc
 import pwndbg.aglib.regs
 import pwndbg.commands
 import pwndbg.rizin
-from pwndbg.color import message
 from pwndbg.commands import CommandCategory
 
 parser = argparse.ArgumentParser(description="Launches rizin.")
@@ -102,10 +101,4 @@ pwndbg> rzpipe pdf @ sym.main
 )
 @pwndbg.commands.OnlyWithFile
 def rzpipe(arguments) -> None:
-    try:
-        rz = pwndbg.rizin.rzpipe()
-        print(rz.cmd(" ".join(arguments)))
-    except ImportError:
-        print(message.error("Could not import rzpipe python library. Is it installed?"))
-    except Exception as e:
-        print(message.error(e))
+    print(pwndbg.rizin.rzcmd(arguments))

pwndbg/ghidra.py

@@ -17,10 +17,9 @@ import pwndbg.rizin
 if pwndbg.dbg.is_gdblib_available():
     import pwndbg.gdblib.symbol
 
-from pwndbg.color import message
 
-r2decompiler = pwndbg.config.add_param(
-    "r2decompiler",
+decompiler = pwndbg.config.add_param(
+    "decompiler",
     "radare2",
     "framework that your ghidra plugin installed",
     param_class=pwndbg.lib.config.PARAM_ENUM,
@@ -28,18 +27,6 @@ r2decompiler = pwndbg.config.add_param(
 )
 
 
-@pwndbg.config.trigger(r2decompiler)
-def set_r2decompiler() -> None:
-    if r2decompiler.value in ["radare2", "rizin"]:
-        return
-    print(
-        message.warn(
-            f"Invalid r2decompiler: `{r2decompiler.value}`, please select from radare2 or rizin"
-        )
-    )
-    r2decompiler.revert_default()
-
-
 def decompile(func=None):
     """
     Return the source of the given function decompiled by ghidra.
@@ -49,22 +36,24 @@ def decompile(func=None):
 
     Raises Exception if any fatal error occurs.
     """
+    func_specified = func is not None
     try:
-        if r2decompiler == "radare2":
-            r2 = pwndbg.radare2.r2pipe()
-            # LD -> list supported decompilers (e cmd.pdc=?)
-            # Outputs for example: pdc\npdg
-            if "pdg" not in r2.cmd("LD").split("\n"):
-                raise Exception("radare2 plugin r2ghidra must be installed and available from r2")
-        else:
-            assert r2decompiler == "rizin"
-            r2 = pwndbg.rizin.rzpipe()
-            # Lc -> list core plugins
-            if "ghidra" not in r2.cmd("Lc"):
-                raise Exception("rizin plugin rzghidra must be installed and available from rz")
+        if decompiler == "radare2":
+            r = pwndbg.radare2.r2pipe()
+        elif decompiler == "rizin":
+            r = pwndbg.rizin.rzpipe()
     except ImportError:
         raise Exception("r2pipe or rzpipe not available, but required for r2/rz->ghidra bridge")
 
+    if pwndbg.aglib.qemu.is_qemu_kernel():
+        pc = pwndbg.aglib.regs[pwndbg.aglib.regs.current.pc]
+        if func is None:
+            func = pwndbg.aglib.symbol.resolve_addr(pc)
+            if func is not None:
+                func = func.split("+")[0]
+        if func is not None:
+            func = f"sym.{func}"
+
     if not func:
         func = (
             hex(pwndbg.aglib.regs[pwndbg.aglib.regs.current.pc])
@@ -72,12 +61,14 @@ def decompile(func=None):
             else "main"
         )
 
-    src = r2.cmdj("pdgj @ " + func)
+    r.cmd(f"afr @ {func}")
+    src = r.cmdj("pdgj @ " + func)
     if not src:
         raise Exception(f"Decompile command failed, check if '{func}' is a valid target")
 
     current_line_marker = "/*%%PWNDBG_CODE_MARKER%%*/"
     source = src.get("code", "")
+    closest_line = 1
 
     # If not running there is no current pc to mark
     if pwndbg.aglib.proc.alive:
@@ -85,6 +76,8 @@ def decompile(func=None):
 
         closest = 0
         for off in (a.get("offset", 0) for a in src.get("annotations", [])):
+            if off == 0 and pc > 0x1000:
+                continue
             if abs(pc - closest) > abs(pc - off):
                 closest = off
         pos_annotations = sorted(
@@ -94,7 +87,7 @@ def decompile(func=None):
 
         # Append code prefix marker for the current line and replace it later
         if pos_annotations:
-            curline = source.count("\n", 0, pos_annotations[0]["start"])
+            curline = closest_line = source.count("\n", 0, pos_annotations[0]["start"])
             source = source.split("\n")
             line = source[curline]
             if line.startswith("    "):
@@ -114,4 +107,16 @@ def decompile(func=None):
 
     # Replace code prefix marker after syntax highlighting
     source = source.replace(current_line_marker, C.prefix(pwndbg.config.code_prefix), 1)
+
+    if not func_specified:
+        source = source.split("\n")
+        n = int(pwndbg.config.context_code_lines)
+
+        # Compute the line range
+        start = max(closest_line - 1 - n // 2, 0)
+        end = min(closest_line - 1 + n // 2 + 1, len(source))
+
+        # split the code
+        source = source[start:end]
+        source = "\n".join(source)
     return source

pwndbg/radare2.py

@@ -6,6 +6,7 @@ from __future__ import annotations
 
 import pwndbg.aglib.elf
 import pwndbg.lib.cache
+from pwndbg.color import message
 
 
 @pwndbg.lib.cache.cache_until("start", "objfile")
@@ -29,9 +30,34 @@ def r2pipe():
 
     import r2pipe
 
+    if pwndbg.aglib.qemu.is_qemu_kernel():
+        flags = ["-e", "bin.cache=true", "-e", "bin.relocs.apply=true"]
+        if (kbase := pwndbg.aglib.kernel.kbase()) and filename == pwndbg.aglib.proc.exe:
+            flags.extend(
+                [
+                    "-e",
+                    "bin.baddr=" + hex(kbase - pwndbg.aglib.elf.get_vmlinux_unrand_base(filename)),
+                ]
+            )
+        r2 = r2pipe.open(filename, flags)
+    else:
         flags = ["-e", "io.cache=true"]
         if pwndbg.aglib.elf.get_elf_info(filename).is_pie and pwndbg.aglib.elf.exe():
             flags.extend(["-B", hex(pwndbg.aglib.elf.exe().address)])
         r2 = r2pipe.open(filename, flags=flags)
-    r2.cmd("aaaa")
+    # LD -> list supported decompilers (e cmd.pdc=?)
+    # Outputs for example: pdc\npdg
+    if "pdg" not in r2.cmd("LD").split("\n"):
+        raise Exception("radare2 plugin r2ghidra must be installed and available from r2")
     return r2
+
+
+def r2cmd(arguments) -> str:
+    try:
+        r2 = r2pipe()
+        return r2.cmd(" ".join(arguments))
+    except ImportError:
+        return message.error("Could not import r2pipe python library. Is it installed?")
+    except Exception as e:
+        return message.error(e)
+    return ""

pwndbg/rizin.py

@@ -7,6 +7,7 @@ from __future__ import annotations
 import pwndbg.aglib.elf
 import pwndbg.dbg
 import pwndbg.lib.cache
+from pwndbg.color import message
 
 
 @pwndbg.lib.cache.cache_until("start", "objfile")
@@ -27,9 +28,33 @@ def rzpipe():
 
     import rzpipe
 
+    if pwndbg.aglib.qemu.is_qemu_kernel():
+        flags = ["-e", "bin.cache=true", "-e", "bin.relocs.apply=true"]
+        if (kbase := pwndbg.aglib.kernel.kbase()) and filename == pwndbg.aglib.proc.exe:
+            flags.extend(
+                [
+                    "-e",
+                    "bin.baddr=" + hex(kbase - pwndbg.aglib.elf.get_vmlinux_unrand_base(filename)),
+                ]
+            )
+        rz = rzpipe.open(filename, flags)
+    else:
         flags = ["-e", "io.cache=true"]
         if pwndbg.aglib.elf.get_elf_info(filename).is_pie and pwndbg.aglib.elf.exe():
             flags.extend(["-B", hex(pwndbg.aglib.elf.exe().address)])
         rz = rzpipe.open(filename, flags=flags)
-    rz.cmd("aaaa")
+    # Lc -> list core plugins
+    if "ghidra" not in rz.cmd("Lc"):
+        raise Exception("rizin plugin rzghidra must be installed and available from rz")
     return rz
+
+
+def rzcmd(arguments) -> str:
+    try:
+        rz = rzpipe()
+        return rz.cmd(" ".join(arguments))
+    except ImportError:
+        return message.error("Could not import rzpipe python library. Is it installed?")
+    except Exception as e:
+        return message.error(e)
+    return ""