Merge remote-tracking branch 'qiling.io/dev' into dev

Author: ucgJhe

qiling/arch/arm.py

@@ -68,7 +68,7 @@ def effective_pc(self) -> int:
         """
 
         # append 1 to pc if in thumb mode, or 0 otherwise
-        return self.regs.pc + int(self.is_thumb)
+        return self.regs.pc | int(self.is_thumb)
 
     @property
     def disassembler(self) -> Cs:

qiling/core.py

@@ -737,8 +737,14 @@ def emu_start(self, begin: int, end: int, timeout: int = 0, count: int = 0):
             count   : max emulation steps (instructions count); unlimited by default
         """
 
-        if self._arch.type in (QL_ARCH.ARM, QL_ARCH.CORTEX_M) and self._arch._init_thumb:
-            begin |= 1
+        # FIXME: we cannot use arch.is_thumb to determine this because unicorn sets the coresponding bit in cpsr
+        # only when pc is set. unicorn sets or clears the thumb mode bit based on pc lsb, ignoring the mode it
+        # was initialized with.
+        #
+        # either unicorn is patched to reflect thumb mode in cpsr upon initialization, or we pursue the same logic
+        # by determining the endianess by address lsb. either way this condition should not be here
+        if getattr(self.arch, '_init_thumb', False):
+            begin |= 0b1
 
         # reset exception status before emulation starts
         self._internal_exception = None

qiling/debugger/gdb/gdb.py

@@ -95,6 +95,11 @@ def __init__(self, ql: Qiling, ip: str = '127.0.0.1', port: int = 9999):
         else:
             entry_point = ql.os.entry_point
 
+        # though linkers set the entry point LSB to indicate arm thumb mode, the
+        # effective entry point address is aligned. make sure we have it aligned
+        if hasattr(ql.arch, 'is_thumb'):
+            entry_point &= ~0b1
+
         # Only part of the binary file will be debugged.
         if ql.entry_point is not None:
             entry_point = ql.entry_point
@@ -234,7 +239,7 @@ def handle_c(subcmd: str) -> Reply:
                 reply = f'S{SIGINT:02x}'
 
             else:
-                if self.ql.arch.regs.arch_pc == self.gdb.last_bp:
+                if getattr(self.ql.arch, 'effective_pc', self.ql.arch.regs.arch_pc) == self.gdb.last_bp:
                     # emulation stopped because it hit a breakpoint
                     reply = f'S{SIGTRAP:02x}'
                 else:
@@ -666,12 +671,6 @@ def handle_s(subcmd: str) -> Reply:
             """Perform a single step.
             """
 
-            # BUG: a known unicorn caching issue causes it to emulate more
-            # steps than requestes. until that issue is fixed, single stepping
-            # is essentially broken.
-            #
-            # @see: https://github.com/unicorn-engine/unicorn/issues/1606
-
             self.gdb.resume_emu(steps=1)
 
             return f'S{SIGTRAP:02x}'
@@ -709,8 +708,9 @@ def handle_Z(subcmd: str) -> Reply:
             #   4 = access watchpoint
 
             if type == 0:
-                self.gdb.bp_insert(addr)
-                return REPLY_OK
+                success = self.gdb.bp_insert(addr, kind)
+
+                return REPLY_OK if success else 'E22'
 
             return REPLY_EMPTY
 
@@ -721,12 +721,9 @@ def handle_z(subcmd: str) -> Reply:
             type, addr, kind = (int(p, 16) for p in subcmd.split(','))
 
             if type == 0:
-                try:
-                    self.gdb.bp_remove(addr)
-                except ValueError:
-                    return 'E22'
-                else:
-                    return REPLY_OK
+                success = self.gdb.bp_remove(addr, kind)
+
+                return REPLY_OK if success else 'E22'
 
             return REPLY_EMPTY
 

qiling/debugger/gdb/utils.py

@@ -18,7 +18,7 @@ def __init__(self, ql: Qiling, entry_point: int, exit_point: int):
         self.ql = ql
 
         self.exit_point = exit_point
-        self.bp_list = []
+        self.swbp = set()
         self.last_bp = None
 
         def __entry_point_hook(ql: Qiling):
@@ -41,32 +41,44 @@ def dbg_hook(self, ql: Qiling, address: int, size: int):
         if address == self.last_bp:
             self.last_bp = None
 
-        elif address in self.bp_list:
+        elif address in self.swbp:
             self.last_bp = address
 
             ql.log.info(f'{PROMPT} breakpoint hit, stopped at {address:#x}')
             ql.stop()
 
-        # # TODO: not sure what this is about
-        # if address + size == self.exit_point:
-        #     ql.log.debug(f'{PROMPT} emulation entrypoint at {self.entry_point:#x}')
-        #     ql.log.debug(f'{PROMPT} emulation exitpoint at {self.exit_point:#x}')
+    def bp_insert(self, addr: int, size: int):
+        targets = set(addr + i for i in range(size or 1))
 
-    def bp_insert(self, addr: int):
-        if addr not in self.bp_list:
-            self.bp_list.append(addr)
-            self.ql.log.info(f'{PROMPT} breakpoint added at {addr:#x}')
+        if targets.intersection(self.swbp):
+            return False
+
+        for bp in targets:
+            self.swbp.add(bp)
+
+        self.ql.log.info(f'{PROMPT} breakpoint added at {addr:#x}')
+
+        return True
+
+    def bp_remove(self, addr: int, size: int) -> bool:
+        targets = set(addr + i for i in range(size or 1))
+
+        if not targets.issubset(self.swbp):
+            return False
+
+        for bp in targets:
+            self.swbp.remove(bp)
 
-    def bp_remove(self, addr: int):
-        self.bp_list.remove(addr)
         self.ql.log.info(f'{PROMPT} breakpoint removed from {addr:#x}')
 
+        return True
+
     def resume_emu(self, address: Optional[int] = None, steps: int = 0):
         if address is None:
             address = self.ql.arch.regs.arch_pc
 
         if getattr(self.ql.arch, 'is_thumb', False):
-            address |= 1
+            address |= 0b1
 
         op = f'stepping {steps} instructions' if steps else 'resuming'
         self.ql.log.info(f'{PROMPT} {op} from {address:#x}')

qiling/debugger/qdb/arch/arch_arm.py

@@ -10,16 +10,21 @@
 class ArchARM(Arch):
     def __init__(self):
         super().__init__()
-
-    @property
-    def regs(self):
-        return (
+        self._regs = (
                 "r0", "r1", "r2", "r3",
                 "r4", "r5", "r6", "r7",
                 "r8", "r9", "r10", "r11",
                 "r12", "sp", "lr", "pc",
                 )
 
+    @property
+    def regs(self):
+        return self._regs
+
+    @regs.setter
+    def regs(self, regs):
+        self._regs += regs
+
     @property
     def regs_need_swapped(self):
         return {

qiling/debugger/qdb/qdb.py

@@ -82,11 +82,7 @@ def bp_handler(ql, address, size, bp_list):
 
         self.cur_addr = self.ql.loader.entry_point
 
-        if self.ql.arch.type == QL_ARCH.CORTEX_M:
-            self._run()
-
-        else:
-            self.init_state = self.ql.save()
+        self.init_state = self.ql.save()
 
         if self._script:
             run_qdb_script(self, self._script)
@@ -118,23 +114,6 @@ def _run(self, address: int = 0, end: int = 0, count: int = 0) -> None:
         if not address:
             address = self.cur_addr
 
-        if self.ql.arch.type == QL_ARCH.CORTEX_M and self.ql.count != 0:
-
-            while self.ql.count:
-
-                if (bp := self.bp_list.pop(self.cur_addr, None)):
-                    if isinstance(bp, TempBreakpoint):
-                        self.del_breakpoint(bp)
-                    else:
-                        qdb_print(QDB_MSG.INFO, f"hit breakpoint at 0x{self.cur_addr:08x}")
-
-                    break
-
-                self.ql.arch.step()
-                self.ql.count -= 1
-
-            return
-
         if getattr(self.ql.arch, 'is_thumb', False):
             address |= 1
 
@@ -227,13 +206,10 @@ def do_step_in(self, *args) -> Optional[bool]:
         if prophecy.where is True:
             return True
 
-        if self.ql.arch == QL_ARCH.CORTEX_M:
-            self.ql.arch.step()
-        else:
-            step = 1
-            # make sure follow branching
-            if prophecy.going is True and self.ql.arch.type == QL_ARCH.MIPS:
-                step += 1
+        step = 1
+        # make sure follow branching
+        if prophecy.going is True and self.ql.arch.type == QL_ARCH.MIPS:
+            step += 1
 
         self._run(count=step)
         self.do_context()

qiling/debugger/qdb/render/render.py

@@ -51,6 +51,7 @@ def wrapper(*args, **kwargs):
     def __init__(self):
         self.regs_a_row = 4
         self.stack_num = 10
+        self.disasm_num = 0x10
         self.color = color
 
     def reg_diff(self, cur_regs, saved_reg_dump):
@@ -206,13 +207,14 @@ def context_asm(self) -> None:
 
         lines = {}
         past_list = []
-        from_addr = self.cur_addr - 0x10
-        to_addr = self.cur_addr + 0x10
+        from_addr = self.cur_addr - self.disasm_num
+        to_addr = self.cur_addr + self.disasm_num
 
         cur_addr = from_addr
         while cur_addr != to_addr:
             insn = self.disasm(cur_addr)
-            cur_addr += self.arch_insn_size
+            # cur_addr += self.arch_insn_size
+            cur_addr += insn.size
             if not insn:
                 continue
             past_list.append(insn)

qiling/debugger/qdb/render/render_arm.py

@@ -16,6 +16,7 @@ class ContextRenderARM(ContextRender, ArchARM):
     def __init__(self, ql, predictor):
         super().__init__(ql, predictor)
         ArchARM.__init__(self)
+        self.disasm_num = 8
 
     @staticmethod
     def print_mode_info(bits):

qiling/os/posix/const.py

@@ -107,6 +107,45 @@
     "IP_MULTICAST_ALL"          : 0x0031,
 }
 
+# https://github.com/torvalds/linux/blob/master/include/uapi/linux/tcp.h
+linux_socket_tcp_options = {
+    "TCP_NODELAY"				: 0x1,
+    "TCP_MAXSEG"				: 0x2,
+    "TCP_CORK"					: 0x3,
+    "TCP_KEEPIDLE"				: 0x4,
+    "TCP_KEEPINTVL"				: 0x5,
+    "TCP_KEEPCNT"				: 0x6,
+    "TCP_SYNCNT"				: 0x7,
+    "TCP_LINGER2"				: 0x8,
+    "TCP_DEFER_ACCEPT"			: 0x9,
+    "TCP_WINDOW_CLAMP"			: 0xa,
+    "TCP_INFO"					: 0xb,
+    "TCP_QUICKACK"				: 0xc,
+    "TCP_CONGESTION"			: 0xd,
+    "TCP_MD5SIG"				: 0xe,
+    "TCP_THIN_LINEAR_TIMEOUTS"	: 0x10,
+    "TCP_THIN_DUPACK"			: 0x11,
+    "TCP_USER_TIMEOUT"			: 0x12,
+    "TCP_REPAIR"				: 0x13,
+    "TCP_REPAIR_QUEUE"			: 0x14,
+    "TCP_QUEUE_SEQ"				: 0x15,
+    "TCP_REPAIR_OPTIONS"		: 0x16,
+    "TCP_FASTOPEN"				: 0x17,
+    "TCP_TIMESTAMP"				: 0x18,
+    "TCP_NOTSENT_LOWAT"			: 0x19,
+    "TCP_CC_INFO"				: 0x1a,
+    "TCP_SAVE_SYN"				: 0x1b,
+    "TCP_SAVED_SYN"				: 0x1c,
+    "TCP_REPAIR_WINDOW"			: 0x1d,
+    "TCP_FASTOPEN_CONNECT"		: 0x1e,
+    "TCP_ULP"					: 0x1f,
+    "TCP_MD5SIG_EXT"			: 0x20,
+    "TCP_FASTOPEN_KEY"			: 0x21,
+    "TCP_FASTOPEN_NO_COOKIE"	: 0x22,
+    "TCP_ZEROCOPY_RECEIVE"		: 0x23,
+    "TCP_INQ"					: 0x24,
+    "TCP_TX_DELAY"				: 0x25,
+}
 
 macos_socket_ip_options = {
     "IP_TOS"                   : 0x0003,
@@ -241,13 +280,15 @@
 linux_arm_socket_options = {
     "SO_DEBUG"           : 0x0001,
     "SO_REUSEADDR"       : 0x0002,
-    "SO_KEEPALIVE"       : 0x0009,
+    "SO_TYPE"            : 0x0003,
+    "SO_ERROR"           : 0x0004,
     "SO_DONTROUTE"       : 0x0005,
     "SO_BROADCAST"       : 0x0006,
-    "SO_LINGER"          : 0x000d,
-    "SO_OOBINLINE"       : 0x000a,
     "SO_SNDBUF"          : 0x0007,
     "SO_RCVBUF"          : 0x0008,
+    "SO_KEEPALIVE"       : 0x0009,
+    "SO_OOBINLINE"       : 0x000a,
+    "SO_LINGER"          : 0x000d,
     "SO_REUSEPORT"       : 0x000f,
     "SO_SNDLOWAT"        : 0x0013,
     "SO_RCVLOWAT"        : 0x0012,
@@ -336,6 +377,7 @@
     "SO_REUSEADDR"              : 0x0004,
     "SO_KEEPALIVE"              : 0x0008,
     "SO_DONTROUTE"              : 0x0010,
+    "SO_BINDTODEVICE"           : 0x0019,
     "SO_BROADCAST"              : 0x0020,
     "SO_LINGER"                 : 0x0080,
     "SO_OOBINLINE"              : 0x0100,