Opportunistic PEP8 and linter-friendly tweaks

Author: elicn

qiling/arch/cortex_m.py

@@ -6,7 +6,7 @@
 from functools import cached_property
 from contextlib import ContextDecorator
 
-from unicorn import Uc, UcError, UC_ARCH_ARM, UC_MODE_ARM, UC_MODE_MCLASS, UC_MODE_THUMB, UC_ERR_OK
+from unicorn import UC_ARCH_ARM, UC_MODE_ARM, UC_MODE_MCLASS, UC_MODE_THUMB
 from capstone import Cs, CS_ARCH_ARM, CS_MODE_ARM, CS_MODE_MCLASS, CS_MODE_THUMB
 from keystone import Ks, KS_ARCH_ARM, KS_MODE_ARM, KS_MODE_THUMB
 

qiling/const.py

@@ -1,15 +1,17 @@
 #!/usr/bin/env python3
-# 
+#
 # Cross Platform and Multi Architecture Advanced Binary Emulation Framework
 #
 
 from enum import Enum, Flag, IntEnum
-from typing import Mapping, Type, TypeVar
+from typing import Final, Mapping, Type, TypeVar
+
 
 class QL_ENDIAN(IntEnum):
     EL = 1
     EB = 2
 
+
 class QL_ARCH(IntEnum):
     X86 = 101
     X8664 = 102
@@ -23,6 +25,7 @@ class QL_ARCH(IntEnum):
     RISCV64 = 111
     PPC = 112
 
+
 class QL_OS(IntEnum):
     LINUX = 201
     FREEBSD = 202
@@ -35,44 +38,53 @@ class QL_OS(IntEnum):
     MCU = 209
     BLOB = 210
 
+
 class QL_VERBOSE(IntEnum):
-    DISABLED = -1 # turn off all the output
-    OFF = 0       # output only warnings 
-    DEFAULT = 1   # output warnings and Qiling execute process information
-    DEBUG = 4     # output all logs above and debug information, include syscall information
-    DISASM = 10   # output all assembly instructions during Qiling execution
-    DUMP = 20     # output any log Qiling can, include instructions and registers
+    DISABLED = -1   # turn off all the output
+    OFF = 0         # output only warnings
+    DEFAULT = 1     # output warnings and Qiling execute process information
+    DEBUG = 4       # output all logs above and debug information, include syscall information
+    DISASM = 10     # output all assembly instructions during Qiling execution
+    DUMP = 20       # output any log Qiling can, include instructions and registers
+
 
 class QL_DEBUGGER(IntEnum):
     GDB = 1
     IDAPRO = 2
     QDB = 3
 
+
 class QL_INTERCEPT(IntEnum):
     CALL = 1
     ENTER = 2
     EXIT = 3
 
+
 class QL_STOP(Flag):
     NONE = 0
     STACK_POINTER = (1 << 0)
-    EXIT_TRAP     = (1 << 1)
+    EXIT_TRAP = (1 << 1)
+
+
+QL_ARCH_INTERPRETER: Final = (QL_ARCH.EVM,)
 
-QL_ARCH_INTERPRETER = (QL_ARCH.EVM,)
+QL_OS_POSIX: Final = (QL_OS.LINUX, QL_OS.FREEBSD, QL_OS.MACOS, QL_OS.QNX)
+QL_OS_BAREMETAL: Final = (QL_OS.MCU,)
 
-QL_OS_POSIX     = (QL_OS.LINUX, QL_OS.FREEBSD, QL_OS.MACOS, QL_OS.QNX)
-QL_OS_BAREMETAL = (QL_OS.MCU,)
 
 QL_HOOK_BLOCK = 0b0001
 QL_CALL_BLOCK = 0b0010
 
 T = TypeVar('T', bound=Enum)
+
+
 def __casefold_enum(e: Type[T]) -> Mapping[str, T]:
     '''Create a casefolded mapping of an enum to allow case-insensitive lookup.
     '''
 
     return dict((k.casefold(), v) for k, v in e._member_map_.items())
 
+
 debugger_map = __casefold_enum(QL_DEBUGGER)
 arch_map     = __casefold_enum(QL_ARCH)
 os_map       = __casefold_enum(QL_OS)

qiling/debugger/gdb/utils.py

@@ -6,13 +6,13 @@
 from typing import Optional
 
 from qiling import Qiling
-from qiling.const import QL_ARCH
 
 # this code is partially based on uDbg
 # @see: https://github.com/iGio90/uDdbg
 
 PROMPT = r'gdb>'
 
+
 class QlGdbUtils:
     def __init__(self, ql: Qiling, entry_point: int, exit_point: int):
         self.ql = ql
@@ -32,7 +32,6 @@ def __entry_point_hook(ql: Qiling):
         # that hook will be used to set up the breakpoint handling hook
         ep_hret = ql.hook_address(__entry_point_hook, entry_point)
 
-
     def dbg_hook(self, ql: Qiling, address: int, size: int):
         if getattr(ql.arch, 'is_thumb', False):
             address |= 1
@@ -53,18 +52,15 @@ def dbg_hook(self, ql: Qiling, address: int, size: int):
         #     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):
         if addr not in self.bp_list:
             self.bp_list.append(addr)
             self.ql.log.info(f'{PROMPT} breakpoint added at {addr:#x}')
 
-
     def bp_remove(self, addr: int):
         self.bp_list.remove(addr)
         self.ql.log.info(f'{PROMPT} breakpoint removed from {addr:#x}')
 
-
     def resume_emu(self, address: Optional[int] = None, steps: int = 0):
         if address is None:
             address = self.ql.arch.regs.arch_pc

qiling/debugger/qdb/qdb.py

@@ -3,12 +3,12 @@
 # Cross Platform and Multi Architecture Advanced Binary Emulation Framework
 #
 
-from typing import Callable, Optional, Mapping, Tuple, Union
-
 import cmd
 
+from typing import Optional, Tuple, Union
+
 from qiling import Qiling
-from qiling.const import QL_OS, QL_ARCH, QL_VERBOSE
+from qiling.const import QL_OS, QL_ARCH
 from qiling.debugger import QlDebugger
 
 from .utils import setup_context_render, setup_branch_predictor, setup_address_marker, SnapshotManager, run_qdb_script
@@ -18,6 +18,7 @@
 
 from .utils import QDB_MSG, qdb_print
 
+
 class QlQdb(cmd.Cmd, QlDebugger):
     """
     The built-in debugger of Qiling Framework
@@ -342,7 +343,7 @@ def do_examine(self, line: str) -> None:
 
         if type(err_msg := self.mm.parse(line)) is str:
             qdb_print(QDB_MSG.ERROR, err_msg)
-    
+
 
     def do_set(self, line: str) -> None:
         """

qiling/os/linux/thread.py

@@ -91,7 +91,7 @@ def __init__(self, ql: Qiling, start_address: int, exit_point: int, context = No
     @property
     def ql(self):
         return self._ql
-    
+
     @ql.setter
     def ql(self, q):
         self._ql = q
@@ -107,23 +107,23 @@ def saved_context(self, ctx):
     @property
     def exit_point(self):
         return self._exit_point
-    
+
     @exit_point.setter
     def exit_point(self, ep):
         self._exit_point = ep
-    
+
     @property
     def start_address(self):
         return self._start_address
-    
+
     @start_address.setter
     def start_address(self, sa):
         self._start_address = sa
 
     @property
     def status(self):
         return self._status
-    
+
     @status.setter
     def status(self, s):
         self._status = s
@@ -150,46 +150,46 @@ def id(self):
 
     def __hash__(self):
         return self.id
-    
+
     def __str__(self):
         return f"[Thread {self.id}]"
 
     @property
     def set_child_tid_address(self):
         return self._set_child_tid_address
-    
+
     @set_child_tid_address.setter
     def set_child_tid_address(self, addr):
         self._set_child_tid_address = addr
-    
+
     @property
     def clear_child_tid_address(self):
         return self._clear_child_tid_address
 
     @clear_child_tid_address.setter
     def clear_child_tid_address(self, addr):
         self._clear_child_tid_address = addr
-    
+
     @property
     def robust_list_head_ptr(self):
         return self._robust_list_head_ptr
-    
+
     @robust_list_head_ptr.setter
     def robust_list_head_ptr(self, p):
         self._robust_list_head_ptr = p
-    
+
     @property
     def robust_list_head_len(self):
         return self._robust_list_head_len
-    
+
     @robust_list_head_len.setter
     def robust_list_head_len(self, l):
         self._robust_list_head_len = l
 
     @property
     def sched_cb(self) -> Callable:
         return self._sched_cb
-    
+
     @sched_cb.setter
     def sched_cb(self, cb):
         self._sched_cb = cb
@@ -207,7 +207,7 @@ def _run(self):
         #    Within both contexts, our program is single thread.
         #
         #    The only fail safe: **Never give up control in Unicorn context.**
-        #    
+        #
         #    In Unicorn context, in other words, in Unicorn callbacks, we do:
         #        - Implement non-blocking syscalls directly.
         #        - Prepare sched_cb for non-unicorn context.
@@ -219,7 +219,7 @@ def _run(self):
         self.ql.arch.regs.arch_pc = self.start_address
         if not self._saved_context:
             self.save()
-        
+
         while self.status != THREAD_STATUS_TERMINATED:
             # Rewrite our status and the current thread.
             self.status = THREAD_STATUS_RUNNING
@@ -235,7 +235,7 @@ def _run(self):
             # Run and log the run event
             start_address = getattr(self.ql.arch, 'effective_pc', self.ql.arch.regs.arch_pc) # For arm thumb.
             self.sched_cb = QlLinuxThread._default_sched_cb
-            
+
             self.ql.log.debug(f"Scheduled from {hex(start_address)}.")
             try:
                 # Known issue for timeout: https://github.com/unicorn-engine/unicorn/issues/1355
@@ -246,7 +246,7 @@ def _run(self):
                 raise e
             self.ql.log.debug(f"Suspended at {hex(self.ql.arch.regs.arch_pc)}")
             self.save()
-            
+
             # Note that this callback may be set by UC callbacks.
             # Some thought on this design:
             #      1. Never give up control during a UC callback.
@@ -275,16 +275,16 @@ def restore(self):
     @abstractmethod
     def set_thread_tls(self, tls_addr):
         pass
-    
+
     @abstractmethod
     def clone(self):
         # This is a workaround to implement our thread based on gevent greenlet.
         # Core idea:
         #     A gevent greenlet can't re-run if it has finished _run method but our framework requires threads to be resumed anytime. Therefore, a workaround is to
         #     use multiple greenlets to represent a single qiling thread.
-        #     
+        #
         #     Of course we can make the greenlet run forever and wait for notifications to resume but that would make the design much more complicated.
-        #     
+        #
         # Caveat:
         #     Don't use thread id to identify the thread object.
         new_thread = self.ql.os.thread_class.spawn(self._ql, self._start_address, self._exit_point, self._saved_context, set_child_tid_addr = None, thread_id = self._thread_id)
@@ -435,7 +435,7 @@ def restore(self):
         self.restore_context()
         self.set_thread_tls(self.tls)
         self.ql.log.debug(f"Restored context: fs={hex(self.ql.arch.regs.fsbase)} tls={hex(self.tls)}")
-    
+
     def clone(self):
         new_thread = super(QlLinuxX8664Thread, self).clone()
         new_thread.tls = self.tls
@@ -458,7 +458,7 @@ def set_thread_tls(self, tls_addr):
     def save(self):
         self.save_context()
         self.tls = self.ql.arch.regs.cp0_userlocal
-        self.ql.log.debug(f"Saved context. cp0={hex(self.ql.arch.regs.cp0_userlocal)}") 
+        self.ql.log.debug(f"Saved context. cp0={hex(self.ql.arch.regs.cp0_userlocal)}")
 
     def restore(self):
         self.restore_context()
@@ -492,7 +492,7 @@ def restore(self):
         self.restore_context()
         self.set_thread_tls(self.tls)
         self.ql.log.debug(f"Restored context. c13_c0_3={hex(self.ql.arch.regs.c13_c0_3)}")
-    
+
     def clone(self):
         new_thread = super(QlLinuxARMThread, self).clone()
         new_thread.tls = self.tls
@@ -519,7 +519,7 @@ def restore(self):
         self.restore_context()
         self.set_thread_tls(self.tls)
         self.ql.log.debug(f"Restored context. tpidr_el0={hex(self.ql.arch.regs.tpidr_el0)}")
-    
+
     def clone(self):
         new_thread = super(QlLinuxARM64Thread, self).clone()
         new_thread.tls = self.tls
@@ -545,7 +545,7 @@ def cur_thread(self, ct):
     @property
     def main_thread(self):
         return self._main_thread
-    
+
     @main_thread.setter
     def main_thread(self, mt):
         self._main_thread = mt
@@ -570,7 +570,7 @@ def _clear_queued_msg(self):
                 self.main_thread.log_file_fd.log(lvl, msg)
         except AttributeError:
             pass
-    
+
     def _prepare_lib_patch(self):
         if self.ql.loader.elf_entry != self.ql.loader.entry_point:
             entry_address = self.ql.loader.elf_entry