Merge pull request #27 from doronz88/improvment/disable_ipc_timeout

Improvment/disable ipc timeout

hilda/hilda_client.py

@@ -12,16 +12,17 @@
 from collections import namedtuple
 from contextlib import contextmanager, suppress
 from datetime import datetime, timezone
-from functools import partial
+from functools import cached_property, partial
 from pathlib import Path
 from typing import Union
 
-import IPython
 import docstring_parser
 import hexdump
+import IPython
 import lldb
 from humanfriendly import prompts
 from humanfriendly.terminal.html import html_to_ansi
+from keystone import Ks, KS_ARCH_ARM64, KS_ARCH_X86, KS_MODE_64, KS_MODE_LITTLE_ENDIAN
 from pygments import highlight
 from pygments.formatters import TerminalTrueColorFormatter
 from pygments.lexers import XmlLexer
@@ -77,7 +78,7 @@ def __init__(self, debugger: lldb.SBDebugger):
         self.breakpoints = {}
         self.captured_objects = {}
         self.registers = Registers(self)
-
+        self.arch = self.target.GetTriple().split('-')[0]
         # should unwind the stack on errors. change this to False in order to debug self-made calls
         # within hilda
         self._evaluation_unwind_on_error = True
@@ -238,6 +239,16 @@ def poke(self, address, buf: bytes):
 
         return retval
 
+    @command()
+    def poke_text(self, address: int, code: str) -> int:
+        """
+        Write instructions to address.
+        :param address:
+        :param code:
+        """
+        bytecode, count = self._ks.asm(code, as_bytes=True)
+        return self.poke(address, bytecode)
+
     @command()
     def peek(self, address, size: int) -> bytes:
         """
@@ -307,15 +318,16 @@ def detach(self):
             self.log_critical('failed to detach')
 
     @command()
-    def disass(self, address, buf, should_print=True) -> lldb.SBInstructionList:
+    def disass(self, address, buf, flavor='intel', should_print=True) -> lldb.SBInstructionList:
         """
         Print disassembly from a given address
+        :param flavor:
         :param address:
         :param buf:
         :param should_print:
         :return:
         """
-        inst = self.target.GetInstructions(lldb.SBAddress(address, self.target), buf)
+        inst = self.target.GetInstructionsWithFlavor(lldb.SBAddress(address, self.target), flavor, buf)
         if should_print:
             print(inst)
         return inst
@@ -593,7 +605,7 @@ def bp(self, address, callback=None, condition: str = None, forced=False, **opti
 
     def bp_callback_router(self, frame, bp_loc, *_):
         """
-        Route the breakpoint callback the the specific breakpoint callback.
+        Route the breakpoint callback the specific breakpoint callback.
         :param lldb.SBFrame frame: LLDB Frame object.
         :param lldb.SBBreakpointLocation bp_loc: LLDB Breakpoint location object.
         """
@@ -887,13 +899,13 @@ def stopped(self, interval=0):
 
         if is_running:
             self.stop()
-            time.sleep(1)
+            time.sleep(interval)
 
         try:
             yield
         finally:
             if is_running:
-                time.sleep(1)
+                time.sleep(interval)
                 self.cont()
 
     @contextmanager
@@ -1101,12 +1113,11 @@ def _serialize_call_params(self, argv):
                 raise NotImplementedError('cannot serialize argument')
         return args_conv
 
-    @staticmethod
-    def _generate_call_expression(address, params):
+    def _generate_call_expression(self, address, params):
         args_type = ','.join(['intptr_t'] * len(params))
         args_conv = ','.join(params)
 
-        if self.target.modules[0].triple.split('-')[0] == 'arm64e':
+        if self.arch == 'arm64e':
             address = f'ptrauth_sign_unauthenticated((void *){address}, ptrauth_key_asia, 0)'
 
         return f'((intptr_t(*)({args_type}))({address}))({args_conv})'
@@ -1163,3 +1174,10 @@ def _monitor_format_value(self, fmt, value):
             return formatters[fmt](value)
         else:
             return f'{value:x} (unsupported format)'
+
+    @cached_property
+    def _ks(self) -> Ks:
+        platforms = {'arm64': Ks(KS_ARCH_ARM64, KS_MODE_LITTLE_ENDIAN),
+                     'arm64e': Ks(KS_ARCH_ARM64, KS_MODE_LITTLE_ENDIAN),
+                     'x86_64h': Ks(KS_ARCH_X86, KS_MODE_64)}
+        return platforms.get(self.arch)

hilda/snippets/mach/CFRunLoopServiceMachPort_hooks.py

@@ -1,45 +1,44 @@
 import lldb
 
-from hilda.exceptions import SymbolAbsentError
 
-
-def _CFRunLoopServiceMachPort_hook(hilda, *args):
-    """
-    :param hilda.hilda_client.HildaClient hilda:
+def disable_mach_msg_errors() -> None:
     """
-    hilda.jump(hilda.CFRunLoopServiceMachPort_while_ea)
-    hilda.cont()
+    Remove the timeout validation from __CFRunLoopServiceMachPort. This is done by patching the mach_msg_timeout_t
+    parameter (4rd one) to MACH_MSG_TIMEOUT_NONE. On arm the parameter passes on `x3` register.
 
+        __int64 __fastcall __CFRunLoopServiceMachPort(
+            mach_port_name_t a1,
+            mach_msg_header_t **a2,
+            mach_port_t *a3,
+            mach_msg_timeout_t a4,
+            voucher_mach_msg_state_t *a5,
+            id *a6)
 
-def disable_mach_msg_errors() -> None:
-    """
-    Disable the error check inside the CFRunLoopServiceMachPort from the mach_msg syscall.
-    This is used to debug slow handling of mach messages.
+        CoreFoundation:__text:0000000186E5012C ___CFRunLoopServiceMachPort
+        CoreFoundation:__text:0000000186E5012C                 SUB             SP, SP, #0x70
+        CoreFoundation:__text:0000000186E50130                 STP             X28, X27, [SP,#0x60+var_50]
+        CoreFoundation:__text:0000000186E50134                 STP             X26, X25, [SP,#0x60+var_40]
+        CoreFoundation:__text:0000000186E50138                 STP             X24, X23, [SP,#0x60+var_30]
+        CoreFoundation:__text:0000000186E5013C                 STP             X22, X21, [SP,#0x60+var_20]
+        CoreFoundation:__text:0000000186E50140                 STP             X20, X19, [SP,#0x60+var_10]
+        CoreFoundation:__text:0000000186E50144                 STP             X29, X30, [SP,#0x60+var_s0]
+        CoreFoundation:__text:0000000186E50148                 ADD             X29, SP, #0x60
+        CoreFoundation:__text:0000000186E5014C                 MOV             X21, X5
+        CoreFoundation:__text:0000000186E50150                 MOV             X22, X4
+        CoreFoundation:__text:0000000186E50154                 MOV             X23, X3       <-------- Timeout parameter
     """
     hilda = lldb.hilda_client
-    with hilda.stopped():
-        instructions = hilda.symbols.__CFRunLoopServiceMachPort.disass(2000, should_print=False)
-        while_ea = None
-        for instruction in instructions:
-            if (while_ea is None) and instruction.DoesBranch():
-                # Beginning of the `while(true) { ... }`
-                while_ea = instruction.GetOperands(hilda.target)
-                hilda.CFRunLoopServiceMachPort_while_ea = int(hilda.file_symbol(eval(while_ea)))
-            elif instruction.GetMnemonic(hilda.target) in ('brk', 'ud2'):
-                symbol = hilda.symbol(instruction.addr.GetLoadAddress(hilda.target))
-                symbol.bp(
-                    _CFRunLoopServiceMachPort_hook,
-                    forced=True,
-                    name=f'__CFRunLoopServiceMachPort-brk-{int(symbol - hilda.symbols.__CFRunLoopServiceMachPort)}'
-                )
+    if hilda.arch == 'x86_64h':
+        return
 
-        # on iOS 16.x, will need to also patch this one
-        try:
-            handle_error = hilda.symbols['__CFRunLoopServiceMachPort.cold.1']
-        except SymbolAbsentError:
-            return
-
-        for instruction in handle_error.disass(2000, should_print=False):
-            if instruction.GetMnemonic(hilda.target) in ('brk', 'ud2'):
-                # mov x0, x0
-                hilda.symbol(instruction.addr.GetLoadAddress(hilda.target)).poke(b'\xe0\x03\x00\xaa')
+    with hilda.stopped():
+        for inst in hilda.symbols.__CFRunLoopServiceMachPort.disass(200, should_print=False):
+            mnemonic = inst.GetMnemonic(hilda.target)
+            operands = inst.GetOperands(hilda.target)
+            if mnemonic != 'mov' or not operands.endswith('x3'):
+                continue
+            addr = inst.GetAddress()
+            file_addr = addr.GetFileAddress()
+            new_inst = f'{mnemonic} {operands.replace("x3", "0")}'
+            hilda.file_symbol(file_addr).poke_text(new_inst)
+            break

hilda/symbol.py

@@ -1,7 +1,8 @@
-from construct import FormatField
-from contextlib import contextmanager
-import struct
 import os
+import struct
+from contextlib import contextmanager
+
+from construct import FormatField
 
 from hilda.objective_c_class import Class
 
@@ -119,6 +120,9 @@ def peek(self, count):
     def poke(self, buf):
         return self._client.poke(self, buf)
 
+    def poke_text(self, code: str) -> int:
+        return self._client.poke_text(self, code)
+
     def peek_str(self):
         return self._client.peek_str(self)