Merge branch 'qilingframework:dev' into dev

Author: xwings

qiling/extensions/trace.py

@@ -2,11 +2,10 @@
 
 # More info, please refer to https://github.com/qilingframework/qiling/pull/765
 
-
 from collections import deque
 from typing import Deque, Iterable, Iterator, Mapping, Tuple
 
-from capstone import Cs, CsInsn, CS_OP_IMM, CS_OP_MEM, CS_OP_REG
+from capstone import Cs, CsInsn, CS_ARCH_X86, CS_OP_IMM, CS_OP_MEM, CS_OP_REG
 from capstone.x86 import X86Op
 from capstone.x86_const import X86_INS_LEA, X86_REG_INVALID, X86_REG_RIP
 
@@ -16,7 +15,7 @@
 
 # <WORKAROUND>
 def __uc2_workaround() -> Mapping[int, int]:
-	"""Starting from Unicron2, Unicron and Capstone Intel registers definitions are
+	"""Starting from Unicorn2, Unicorn and Capstone Intel registers definitions are
 	no longer aligned and cannot be used interchangebly. This temporary workaround
 	maps capstone x86 registers definitions to unicorn x86 registers definitions.
 
@@ -47,6 +46,7 @@ def __get_trace_records(ql: Qiling, address: int, size: int, md: Cs) -> Iterator
 	# unicorn denotes unsupported instructions by a magic size value. though these instructions
 	# are not emulated, capstone can still parse them.
 	if size == 0xf1f1f1f1:
+		# note that invalid instructions will generate a StopIteration exception here
 		yield next(__get_trace_records(ql, address, 16, md))
 		return
 
@@ -125,6 +125,7 @@ def __parse_op(op: X86Op) -> str:
 					2: 'word',
 					4: 'dword',
 					8: 'qword',
+					10: 'fword',
 					16: 'xmmword'
 				}[op.size]
 
@@ -154,13 +155,15 @@ def enable_full_trace(ql: Qiling):
 	md = ql.create_disassembler()
 	md.detail = True
 
+	assert md.arch == CS_ARCH_X86, 'currently available only for intel architecture'
+
 	# if available, use symbols map to resolve memory accesses
 	symsmap = getattr(ql.loader, 'symsmap', {})
 
 	# show trace lines in a darker color so they would be easily distinguished from
 	# ordinary log records
-	DarkGray = "\x1b[90m"
-	Default = "\x1b[39m"
+	faded_color = "\033[2m"
+	reset_color = "\033[0m"
 
 	def __trace_hook(ql: Qiling, address: int, size: int):
 		"""[internal] Trace hook callback.
@@ -169,7 +172,7 @@ def __trace_hook(ql: Qiling, address: int, size: int):
 		for record in __get_trace_records(ql, address, size, md):
 			line = __to_trace_line(record, symsmap)
 
-			ql.log.debug(f'{DarkGray}{line}{Default}')
+			ql.log.debug(f'{faded_color}{line}{reset_color}')
 
 	ql.hook_code(__trace_hook)
 
@@ -189,6 +192,8 @@ def enable_history_trace(ql: Qiling, nrecords: int):
 	md = ql.create_disassembler()
 	md.detail = True
 
+	assert md.arch == CS_ARCH_X86, 'currently available only for intel architecture'
+
 	# if available, use symbols map to resolve memory accesses
 	symsmap = getattr(ql.loader, 'symsmap', {})
 

qiling/os/fcall.py

@@ -114,7 +114,7 @@ def __get_typed_args(proto: Mapping[str, Any], args: Mapping[str, Any]) -> Itera
 		if len(names) > len(types):
 			types.extend([None] * (len(names) - len(types)))
 
-		return zip(types, names, values)
+		return tuple(zip(types, names, values))
 
 	def call(self, func: CallHook, proto: Mapping[str, Any], params: Mapping[str, Any], hook_onenter: Optional[OnEnterHook], hook_onexit: Optional[OnExitHook], passthru: bool) -> Tuple[Iterable[TypedArg], int, int]:
 		"""Execute a hooked function.

qiling/os/posix/const.py

@@ -533,6 +533,24 @@
     'O_LARGEFILE': None
 }
 
+windows_x86_open_flags = {
+    'O_RDONLY': 0x0,
+    'O_WRONLY': 0x1,
+    'O_RDWR': 0x2,
+    'O_NONBLOCK': None,
+    'O_APPEND': 0x8,
+    'O_ASYNC': None,
+    'O_SYNC': None,
+    'O_NOFOLLOW': None,
+    'O_CREAT': 0x100,
+    'O_TRUNC': 0x200,
+    'O_EXCL': 0x400,
+    'O_NOCTTY': None,
+    'O_DIRECTORY': None,
+    'O_BINARY': 0x8000,
+    'O_LARGEFILE': None
+}
+
 qnx_arm64_open_flags = {
     'O_RDONLY'    : 0x00000,
     'O_WRONLY'    : 0x00001,

qiling/os/posix/const_mapping.py

@@ -64,6 +64,8 @@ def flag_mapping(flags, mapping_name, mapping_from, mapping_to):
             f = macos_x86_open_flags
     elif ql.ostype == QL_OS.FREEBSD:
         f = freebsd_x86_open_flags
+    elif ql.ostype == QL_OS.WINDOWS:
+        f = windows_x86_open_flags
     elif ql.ostype == QL_OS.QNX:
         f = qnx_arm64_open_flags
 
@@ -73,6 +75,8 @@ def flag_mapping(flags, mapping_name, mapping_from, mapping_to):
         t = macos_x86_open_flags
     elif ql.platform_os == QL_OS.FREEBSD:
         t = freebsd_x86_open_flags
+    elif ql.platform_os == QL_OS.WINDOWS:
+        t = windows_x86_open_flags
 
     if f == t:
         return flags

qiling/os/utils.py

@@ -68,21 +68,21 @@ def string_appearance(self, s: str) -> None:
             self.appeared_strings.setdefault(token, set()).add(self.syscalls_counter)
 
     @staticmethod
-    def read_string(ql: Qiling, address: int, terminator: str) -> str:
-        result = ""
+    def read_string(ql: Qiling, address: int, terminator: bytes) -> str:
+        result = bytearray()
         charlen = len(terminator)
 
         char = ql.mem.read(address, charlen)
 
-        while char.decode(errors="ignore") != terminator:
+        while char != terminator:
             address += charlen
-            result += char.decode(errors="ignore")
+            result += char
             char = ql.mem.read(address, charlen)
 
-        return result
+        return result.decode(errors="ignore")
 
     def read_wstring(self, address: int) -> str:
-        s = QlOsUtils.read_string(self.ql, address, '\x00\x00')
+        s = QlOsUtils.read_string(self.ql, address, b'\x00\x00')
 
         # We need to remove \x00 inside the string. Compares do not work otherwise
         s = s.replace("\x00", "")
@@ -91,7 +91,7 @@ def read_wstring(self, address: int) -> str:
         return s
 
     def read_cstring(self, address: int) -> str:
-        s = QlOsUtils.read_string(self.ql, address, '\x00')
+        s = QlOsUtils.read_string(self.ql, address, b'\x00')
 
         self.string_appearance(s)
 
@@ -184,24 +184,3 @@ def printf(self, format: str, args: MutableSequence, wstring: bool = False) -> i
 
     def update_ellipsis(self, params: MutableMapping, args: Sequence) -> None:
         params.update((f'{QlOsUtils.ELLIPSIS_PREF}{i}', a) for i, a in enumerate(args))
-
-    def exec_arbitrary(self, start: int, end: int):
-        old_sp = self.ql.reg.arch_sp
-
-        # we read where this hook is supposed to return
-        ret = self.ql.stack_read(0)
-
-        def restore(ql: Qiling):
-            self.ql.log.debug(f"Executed code from {start:#x} to {end:#x}")
-            # now we can restore the register to be where we were supposed to
-            ql.reg.arch_sp = old_sp + ql.pointersize
-            ql.reg.arch_pc = ret
-
-            # we want to execute the code once, not more
-            hret.remove()
-
-        # we have to set an address to restore the registers
-        hret = self.ql.hook_address(restore, end)
-        # we want to rewrite the return address to the function
-        self.ql.stack_write(0, start)
-

qiling/os/windows/dlls/kernel32/errhandlingapi.py

@@ -75,9 +75,7 @@ def hook_SetErrorMode(ql: Qiling, address: int, params):
 def hook_RaiseException(ql: Qiling, address: int, params):
     func_addr = ql.os.handle_manager.search("TopLevelExceptionHandler").obj
 
-    # TODO: this implementation won't work most of the time
-    size = find_size_function(ql, func_addr)
-    ql.os.exec_arbitrary(func_addr, func_addr + size)
+    ql.os.fcall.call_native(func_addr, [], None)
 
     return 0
 
@@ -122,7 +120,6 @@ def exec_standard_into(ql: Qiling, intno: int, user_data):
         ql.reg.esi = user_data
 
     addr = params["Handler"]
-    #size = find_size_function(ql, addr)
 
     # the interrupts 0x2d, 0x3 must be hooked
     hook = ql.hook_intno(exec_standard_into, 0x3, user_data=addr)

qiling/os/windows/utils.py

@@ -37,21 +37,6 @@ def path_leaf(path):
     head, tail = ntpath.split(path)
     return tail or ntpath.basename(head)
 
-# FIXME: determining a function size by locating 'ret' opcodes in its code is a very unreliable way, to say
-# the least. not only that 'ret' instructions may appear more than once in a single function, they not are
-# necessarily located at the last function basic block: think of a typical nested loop spaghetty.
-#
-# also, there is no telling whether a 0xC3 value found in function code is actually a 'ret' instruction, or
-# just part of a magic value (e.g. "mov eax, 0xffffffc3").
-#
-# finally, if this method happens to find the correct function size, by any chance, that would be a pure luck.
-def find_size_function(ql: Qiling, func_addr: int):
-    # We have to retrieve the return address position
-    code = ql.mem.read(func_addr, 0x100)
-    return_procedures = [b"\xc3", b"\xc2", b"\xcb", b"\xca"]
-    min_index = min([code.index(return_value) for return_value in return_procedures if return_value in code])
-    return min_index
-
 
 def io_Write(ql: Qiling, in_buffer: bytes):
     heap = ql.os.heap