Merge pull request #911 from elicn/posix-improv

Refactored POSIX syscalls

Author: xwings

qiling/core.py

@@ -178,7 +178,6 @@ def __init__(
                                                               self._log_file,
                                                               self._console,
                                                               self._filter,
-                                                              self._multithread,
                                                               self._log_override,
                                                               self._log_plain)
 

qiling/os/posix/posix.py

@@ -161,36 +161,40 @@ def getNameFromErrorCode(ret: int) -> str:
     def load_syscall(self):
         # import syscall mapping function
         map_syscall = ql_syscall_mapping_function(self.ql.ostype)
-        syscall = self.syscall
-        syscall_name = map_syscall(self.ql, syscall)
+        syscall_id = self.syscall
+        syscall_name = map_syscall(self.ql, syscall_id)
 
         # get syscall on-enter hook (if any)
         hooks_dict = self.posix_syscall_hooks[QL_INTERCEPT.ENTER]
-        onenter_hook = hooks_dict.get(syscall_name) or hooks_dict.get(syscall)
+        onenter_hook = hooks_dict.get(syscall_name) or hooks_dict.get(syscall_id)
 
         # get syscall on-exit hook (if any)
         hooks_dict = self.posix_syscall_hooks[QL_INTERCEPT.EXIT]
-        onexit_hook = hooks_dict.get(syscall_name) or hooks_dict.get(syscall)
+        onexit_hook = hooks_dict.get(syscall_name) or hooks_dict.get(syscall_id)
 
         # get syscall replacement hook (if any)
         hooks_dict = self.posix_syscall_hooks[QL_INTERCEPT.CALL]
-        syscall_hook = hooks_dict.get(syscall_name) or hooks_dict.get(syscall)
+        syscall_hook = hooks_dict.get(syscall_name) or hooks_dict.get(syscall_id)
+
+        if not syscall_hook:
+            osname = ostype_convert_str(self.ql.ostype)
+            os_syscalls = ql_get_module_function(f"qiling.os.{osname.lower()}", "syscall")
+            posix_syscalls = ql_get_module_function(f"qiling.os.posix", "syscall")
+
+            # look in os-specific and posix syscall hooks
+            syscall_hook = getattr(os_syscalls, syscall_name, None) or getattr(posix_syscalls, syscall_name, None)
 
         if syscall_hook:
             syscall_name = syscall_hook.__name__
-        else:
-            _ostype_str = ostype_convert_str(self.ql.ostype)
-            _posix_syscall = ql_get_module_function(f"qiling.os.posix", "syscall")
-            _os_syscall = ql_get_module_function(f"qiling.os.{_ostype_str.lower()}", "syscall")
 
-            if syscall_name in dir(_posix_syscall) or syscall_name in dir(_os_syscall):
-                syscall_hook = eval(syscall_name)
-                syscall_name = syscall_hook.__name__
-            else:
-                syscall_hook = None
+            # extract the parameters list from hook signature
+            param_names = tuple(signature(syscall_hook).parameters.values())
 
-        if syscall_hook:
-            params = [self.__syscall_cc.getRawParam(i) for i in range(6)]
+            # skip first arg (always 'ql') and filter out python special args (*args and **kwargs)
+            param_names = [info.name for info in param_names[1:] if info.kind == Parameter.POSITIONAL_OR_KEYWORD]
+
+            # read parameter values
+            params = [self.__syscall_cc.getRawParam(i) for i in range(len(param_names))]
 
             try:
         		# if set, fire up the on-enter hook and let it override original args set
@@ -218,24 +222,14 @@ def load_syscall(self):
                 raise
 
             except Exception as e:
-                self.ql.log.exception("")
-                self.ql.log.info(f'Syscall ERROR: {syscall_name} DEBUG: {e}')
+                self.ql.log.exception(f'Syscall ERROR: {syscall_name} DEBUG: {e}')
                 raise e
 
             # print out log entry
-            syscall_basename = syscall_hook.__name__[len(SYSCALL_PREF):]
+            syscall_basename = syscall_name[len(SYSCALL_PREF):]
             args = []
 
-            # ignore first arg, which is 'ql'
-            args_info = tuple(signature(syscall_hook).parameters.values())[1:]
-
-            for info, value in zip(args_info, params):
-                # skip python special args, like: *args and **kwargs
-                if info.kind != Parameter.POSITIONAL_OR_KEYWORD:
-                    continue
-
-                name = info.name
-
+            for name, value in zip(param_names, params):
                 # cut the first part of the arg if it is of form fstatat64_fd
                 if name.startswith(f'{syscall_basename}_'):
                     name = name.partition('_')[-1]
@@ -247,7 +241,7 @@ def load_syscall(self):
 
             # record syscall statistics
             self.utils.syscalls.setdefault(syscall_name, []).append({
-                "params": dict(zip((f'param{i}' for i in range(6)), params)),
+                "params": dict(zip(param_names, params)),
                 "result": retval,
                 "address": self.ql.reg.arch_pc,
                 "return_address": None,
@@ -256,10 +250,10 @@ def load_syscall(self):
 
             self.utils.syscalls_counter += 1
         else:
-            self.ql.log.warning(f'{self.ql.reg.arch_pc:#x}: syscall {syscall_name} number = {syscall:#x}({syscall:d}) not implemented')
+            self.ql.log.warning(f'{self.ql.reg.arch_pc:#x}: syscall {syscall_name} number = {syscall_id:#x}({syscall_id:d}) not implemented')
 
             if self.ql.debug_stop:
-                raise QlErrorSyscallNotFound("Syscall Not Found")
+                raise QlErrorSyscallNotFound(f'Syscall not found: {syscall_name}')
 
     def get_syscall(self) -> int:
         if self.ql.archtype == QL_ARCH.ARM:

qiling/os/posix/stat.py

@@ -6,32 +6,30 @@
 import os
 
 class StatBase:
-    def __init__(self):
-        self._stat_buf = None
+    def __init__(self, stat: os.stat_result):
+        self._stat_buf = stat
 
     # Never iterate this object!
     def __getitem__(self, key):
         if type(key) is not str:
             raise TypeError
-        if not key.startswith("__") and key in dir(self._stat_buf):
+
+        if not key.startswith("__") and hasattr(self._stat_buf, key):
             return self._stat_buf.__getattribute__(key)
+
         return 0
-    
+
     def __getattr__(self, key):
         return self.__getitem__(key)
+
 class Stat(StatBase):
     def __init__(self, path):
-        super(Stat, self).__init__()
-        self._stat_buf = os.stat(path)
+        super().__init__(os.stat(path))
 
 class Fstat(StatBase):
-    def __init__(self, fd):
-        super(Fstat, self).__init__()
-        self._stat_buf = os.fstat(fd)
+    def __init__(self, fd: int):
+        super().__init__(os.fstat(fd))
 
 class Lstat(StatBase):
     def __init__(self, path):
-        super(Lstat, self).__init__()
-        self._stat_buf = os.lstat(path)
-
-
+        super().__init__(os.lstat(path))

qiling/os/posix/syscall/fcntl.py

@@ -3,16 +3,16 @@
 # Cross Platform and Multi Architecture Advanced Binary Emulation Framework
 #
 
+import os
+
 from qiling import Qiling
-from qiling.const import *
-from qiling.os.linux.thread import *
-from qiling.os.posix.filestruct import *
-from qiling.os.filestruct import *
+from qiling.const import QL_OS, QL_ARCH
+from qiling.exception import QlSyscallError
 from qiling.os.posix.const import *
-from qiling.os.posix.const_mapping import *
-from qiling.exception import *
+from qiling.os.posix.const_mapping import ql_open_flag_mapping, open_flags_mapping
+from qiling.os.posix.filestruct import ql_socket
 
-def ql_syscall_open(ql: Qiling, filename, flags, mode, *args, **kw):
+def ql_syscall_open(ql: Qiling, filename: int, flags: int, mode: int):
     path = ql.os.utils.read_cstring(filename)
     real_path = ql.os.path.transform_to_real_path(path)
     relative_path = ql.os.path.transform_to_relative_path(path)
@@ -38,12 +38,13 @@ def ql_syscall_open(ql: Qiling, filename, flags, mode, *args, **kw):
     ql.log.debug("open(%s, %s, 0o%o) = %d" % (relative_path, open_flags_mapping(flags, ql.archtype), mode, regreturn))
 
     if regreturn >= 0 and regreturn != 2:
-        ql.log.debug("File Found: %s" % real_path)
+        ql.log.debug(f'File found: {real_path:s}')
     else:
-        ql.log.debug("File Not Found %s" % real_path)
+        ql.log.debug(f'File not found {real_path:s}')
+
     return regreturn
 
-def ql_syscall_creat(ql: Qiling, filename, mode, *args, **kw):
+def ql_syscall_creat(ql: Qiling, filename: int, mode: int):
     flags = linux_open_flags["O_WRONLY"] | linux_open_flags["O_CREAT"] | linux_open_flags["O_TRUNC"]
 
     path = ql.os.utils.read_cstring(filename)
@@ -71,13 +72,14 @@ def ql_syscall_creat(ql: Qiling, filename, mode, *args, **kw):
     ql.log.debug("creat(%s, %s, 0o%o) = %d" % (relative_path, open_flags_mapping(flags, ql.archtype), mode, regreturn))
 
     if regreturn >= 0 and regreturn != 2:
-        ql.log.debug("File Found: %s" % real_path)
+        ql.log.debug(f'File found: {real_path:s}')
     else:
-        ql.log.debug("File Not Found %s" % real_path)
+        ql.log.debug(f'File not found {real_path:s}')
+
     return regreturn
 
-def ql_syscall_openat(ql: Qiling, fd, path, flags, mode, *args, **kw):
-    path = ql.os.utils.read_cstring(path)
+def ql_syscall_openat(ql: Qiling, fd: int, path: int, flags: int, mode: int):
+    file_path = ql.os.utils.read_cstring(path)
     # real_path = ql.os.path.transform_to_real_path(path)
     # relative_path = ql.os.path.transform_to_relative_path(path)
 
@@ -99,28 +101,27 @@ def ql_syscall_openat(ql: Qiling, fd, path, flags, mode, *args, **kw):
             except:
                 dir_fd = None
 
-            ql.os.fd[idx] = ql.os.fs_mapper.open_ql_file(path, flags, mode, dir_fd)
+            ql.os.fd[idx] = ql.os.fs_mapper.open_ql_file(file_path, flags, mode, dir_fd)
             regreturn = idx
         except QlSyscallError as e:
             regreturn = -e.errno
 
-    ql.log.debug(f'openat(fd = {fd:d}, path = {path}, flags = {open_flags_mapping(flags, ql.archtype)}, mode = {mode:#o}) = {regreturn:d}')
+    ql.log.debug(f'openat(fd = {fd:d}, path = {file_path}, flags = {open_flags_mapping(flags, ql.archtype)}, mode = {mode:#o}) = {regreturn:d}')
 
     return regreturn
 
 
-def ql_syscall_fcntl(ql: Qiling, fcntl_fd, fcntl_cmd, fcntl_arg, *args, **kw):
-    if not (0 <= fcntl_fd < NR_OPEN) or \
-            ql.os.fd[fcntl_fd] == 0:
+def ql_syscall_fcntl(ql: Qiling, fd: int, cmd: int, arg: int):
+    if not (0 <= fd < NR_OPEN) or ql.os.fd[fd] == 0:
         return -EBADF
 
-    f = ql.os.fd[fcntl_fd]
-    
-    if fcntl_cmd == F_DUPFD:
-        if 0 <= fcntl_arg < NR_OPEN:            
+    f = ql.os.fd[fd]
+
+    if cmd == F_DUPFD:
+        if 0 <= arg < NR_OPEN:
             for idx, val in enumerate(ql.os.fd):
-                if val == 0 and idx >= fcntl_arg:
-                    new_fd = ql.os.fd[fcntl_fd].dup()
+                if val == 0 and idx >= arg:
+                    new_fd = ql.os.fd[fd].dup()
                     ql.os.fd[idx] = new_fd
                     regreturn = idx
                     break
@@ -129,18 +130,18 @@ def ql_syscall_fcntl(ql: Qiling, fcntl_fd, fcntl_cmd, fcntl_arg, *args, **kw):
         else:
             regreturn = -EINVAL
 
-    elif fcntl_cmd == F_GETFD:
+    elif cmd == F_GETFD:
         regreturn = getattr(f, "close_on_exec", 0)
 
-    elif fcntl_cmd == F_SETFD:
-        f.close_on_exec = 1 if fcntl_arg & FD_CLOEXEC else 0
+    elif cmd == F_SETFD:
+        f.close_on_exec = 1 if arg & FD_CLOEXEC else 0
         regreturn = 0
 
-    elif fcntl_cmd == F_GETFL:
-        regreturn = ql.os.fd[fcntl_fd].fcntl(fcntl_cmd, fcntl_arg)        
+    elif cmd == F_GETFL:
+        regreturn = ql.os.fd[fd].fcntl(cmd, arg)
 
-    elif fcntl_cmd == F_SETFL:
-        ql.os.fd[fcntl_fd].fcntl(fcntl_cmd, fcntl_arg)
+    elif cmd == F_SETFL:
+        ql.os.fd[fd].fcntl(cmd, arg)
         regreturn = 0
 
     else:
@@ -149,45 +150,50 @@ def ql_syscall_fcntl(ql: Qiling, fcntl_fd, fcntl_cmd, fcntl_arg, *args, **kw):
     return regreturn
 
 
-def ql_syscall_fcntl64(ql: Qiling, fcntl_fd, fcntl_cmd, fcntl_arg, *args, **kw):
+def ql_syscall_fcntl64(ql: Qiling, fd: int, cmd: int, arg: int):
 
     # https://linux.die.net/man/2/fcntl64
-    if fcntl_cmd == F_DUPFD:
-        if 0 <= fcntl_arg < NR_OPEN and 0 <= fcntl_fd < NR_OPEN:
-            if ql.os.fd[fcntl_fd] != 0:
-                new_fd = ql.os.fd[fcntl_fd].dup()
+    if cmd == F_DUPFD:
+        if 0 <= arg < NR_OPEN and 0 <= fd < NR_OPEN:
+            if ql.os.fd[fd] != 0:
+                new_fd = ql.os.fd[fd].dup()
                 for idx, val in enumerate(ql.os.fd):
-                    if val == 0 and idx >= fcntl_arg:
+                    if val == 0 and idx >= arg:
                         ql.os.fd[idx] = new_fd
                         regreturn = idx
                         break
             else:
                 regreturn = -1
         else:
             regreturn = -1
-    elif fcntl_cmd == F_GETFL:
+
+    elif cmd == F_GETFL:
         regreturn = 2
-    elif fcntl_cmd == F_SETFL:
-        if isinstance(ql.os.fd[fcntl_fd], ql_socket):
-            ql.os.fd[fcntl_fd].fcntl(fcntl_cmd, fcntl_arg)
+
+    elif cmd == F_SETFL:
+        if isinstance(ql.os.fd[fd], ql_socket):
+            ql.os.fd[fd].fcntl(cmd, arg)
         regreturn = 0
-    elif fcntl_cmd == F_GETFD:
+
+    elif cmd == F_GETFD:
         regreturn = 2
-    elif fcntl_cmd == F_SETFD:
+
+    elif cmd == F_SETFD:
         regreturn = 0
+
     else:
         regreturn = 0
 
     return regreturn
 
 
-def ql_syscall_flock(ql, flock_fd, flock_operation, *args, **kw):
+def ql_syscall_flock(ql: Qiling, fd: int, operation: int):
     # Should always return 0, we don't need a actual file lock
-    regreturn = 0
-    return regreturn
 
+    return 0
 
-def ql_syscall_rename(ql: Qiling, oldname_buf, newname_buf, *args, **kw):
+
+def ql_syscall_rename(ql: Qiling, oldname_buf: int, newname_buf: int):
     """
     rename(const char *oldpath, const char *newpath)
     description: change the name or location of a file
@@ -208,10 +214,8 @@ def ql_syscall_rename(ql: Qiling, oldname_buf, newname_buf, *args, **kw):
 
     try:
         os.rename(old_realpath, new_realpath)
-        regreturn = 0
     except OSError:
-        ql.log.exception(f"rename(): {newpath} is exist!")
+        ql.log.exception(f"rename(): {newpath} exists!")
         regreturn = -1
 
     return regreturn
-