Merge pull request #1050 from kabeor/dev

add QNX set_api() support

Author: xwings

.github/workflows/build-ci.yml

@@ -78,9 +78,6 @@ jobs:
         wget https://github.com/qilingframework/rootfs/archive/refs/heads/master.zip
         unzip master.zip && mv rootfs-master rootfs
         cd ../qiling
-        rm -rf engine
-        wget https://github.com/qilingframework/engine/archive/refs/heads/main.zip
-        unzip main.zip && mv engine-main engine
         cd ../examples/rootfs/x86_linux/kernel && unzip -P infected m0hamed_rootkit.ko.zip
         cd ../../../../
         pip3 install -e .[evm]

examples/hello_arm_qnx_customapi.py

@@ -0,0 +1,34 @@
+#!/usr/bin/env python3
+#
+# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
+#
+
+import sys
+sys.path.append("..")
+
+from qiling import Qiling
+from qiling.const import QL_INTERCEPT, QL_CALL_BLOCK, QL_VERBOSE
+from qiling.os.const import STRING
+
+def my_puts_onenter(ql: Qiling):
+    params = ql.os.resolve_fcall_params({'s': STRING})
+
+    print(f'puts("{params["s"]}")')
+    return QL_CALL_BLOCK
+
+def my_printf_onenter(ql: Qiling):
+    params = ql.os.resolve_fcall_params({'s': STRING})
+
+    print(f'printf("{params["s"]}")')
+    return QL_CALL_BLOCK
+
+def my_puts_onexit(ql: Qiling):
+    print(f'after puts')
+    return QL_CALL_BLOCK
+
+if __name__ == "__main__":
+    ql = Qiling(["rootfs/arm_qnx/bin/hello_static"], "rootfs/arm_qnx")
+    ql.set_api('puts', my_puts_onenter, QL_INTERCEPT.ENTER)
+    ql.set_api('printf', my_printf_onenter, QL_INTERCEPT.ENTER)
+    ql.set_api('puts', my_puts_onexit, QL_INTERCEPT.EXIT)
+    ql.run()

qiling/os/qnx/message.py

@@ -98,7 +98,7 @@ def ql_qnx_msg_io_connect(ql:Qiling, coid, smsg, sparts, rmsg, rparts, *args, **
         else:
             # TODO: Can we throw this exception here?
             # raise NotImplementedError(f'msg_io_connect(_IO_CONNECT_COMBINE) for type 0x{x_type:x} not implemented')
-            ql.log.warn(f'msg_io_connect(_IO_CONNECT_COMBINE) for type 0x{x_type:x} not implemented')
+            ql.log.warning(f'msg_io_connect(_IO_CONNECT_COMBINE) for type 0x{x_type:x} not implemented')
     elif subtype == 2: # == _IO_CONNECT_OPEN
         ql.log.debug(f'open(path = {path}, openflags = 0x{ioflag:x}, openmode = 0x{real_mode:x})')
         ql.os.connections[coid].fd = ql_syscall_open(ql, ql.unpack32(ql.mem.read(smsg + 8, 4)), ioflag, real_mode)
@@ -127,7 +127,7 @@ def ql_qnx_msg_io_connect(ql:Qiling, coid, smsg, sparts, rmsg, rparts, *args, **
     elif os.path.isfile(real_path):
         umask = file_stats['_S_IFREG']
     else:
-        ql.log.warn("msg_io_connect(): type of {real_path} not handled properly?")
+        ql.log.warning("msg_io_connect(): type of {real_path} not handled properly?")
         umask = 0
     # struct _io_connect_link_reply in lib/c/public/sys/iomsg.h
     ql.mem.write(iov_base, pack("<IIBBHIHH", 0, file_type, eflag, 0, 0, umask, 0, 0))
@@ -190,7 +190,7 @@ def ql_qnx_msg_mem_ctrl(ql:Qiling, coid, smsg, sparts, rmsg, rparts, *args, **kw
     if not subtype in mem_ctrl_subtypes:
         raise NotImplementedError(f'MEM_CTRL subtype {subtype} not implemented')
 
-    ql.log.warn(f'msg_mem_ctrl(subtype = {mem_ctrl_subtypes[subtype]}, flags = 0x{flags:x}, addr = 0x{addr:x}, len = 0x{len:x}) not implemented')
+    ql.log.warning(f'msg_mem_ctrl(subtype = {mem_ctrl_subtypes[subtype]}, flags = 0x{flags:x}, addr = 0x{addr:x}, len = 0x{len:x}) not implemented')
     # TODO: implement mem_ctrl
     return -1
 

qiling/os/qnx/qnx.py

@@ -3,19 +3,34 @@
 # Cross Platform and Multi Architecture Advanced Binary Emulation Framework
 #
 
+from typing import Callable
 import os
 from unicorn import UcError
 
 from qiling.os.posix.posix import QlOsPosix
 from qiling.os.qnx.const import NTO_SIDE_CHANNEL, SYSMGR_PID, SYSMGR_CHID, SYSMGR_COID
 from qiling.os.qnx.helpers import QnxConn
 from qiling.os.qnx.structs import _thread_local_storage
-from qiling.const import QL_ARCH
+from qiling.os.fcall import QlFunctionCall
+from qiling.cc import QlCC, intel, arm, mips, riscv
+from qiling.const import QL_ARCH, QL_INTERCEPT
 
 class QlOsQnx(QlOsPosix):
     def __init__(self, ql):
         super(QlOsQnx, self).__init__(ql)
         self.load()
+        
+        cc: QlCC = {
+            QL_ARCH.X86   : intel.cdecl,
+            QL_ARCH.X8664 : intel.amd64,
+            QL_ARCH.ARM   : arm.aarch32,
+            QL_ARCH.ARM64 : arm.aarch64,
+            QL_ARCH.MIPS  : mips.mipso32,
+            QL_ARCH.RISCV : riscv.riscv,
+            QL_ARCH.RISCV64: riscv.riscv,
+        }[ql.archtype](ql)
+
+        self.fcall = QlFunctionCall(ql, cc)
 
         # use counters to get free Ids
         self.channel_id = 1
@@ -44,6 +59,10 @@ def hook_syscall(self, intno= None, int = None):
         return self.load_syscall()
 
 
+    def add_function_hook(self, fn: str, cb: Callable, intercept: QL_INTERCEPT):
+        self.ql.os.function_hook.add_function_hook(fn, cb, intercept)
+
+
     def hook_sigtrap(self, intno= None, int = None):
         self.ql.log.info("Trap Found")
         self.emu_error()

tests/test_qnx.py

@@ -9,6 +9,8 @@
 from qiling import *
 from qiling.exception import *
 from qiling.const import QL_VERBOSE
+from qiling.const import QL_INTERCEPT, QL_CALL_BLOCK, QL_VERBOSE
+from qiling.os.const import STRING
 
 class QNXTest(unittest.TestCase):
     def test_arm_qnx_static(self):
@@ -20,7 +22,56 @@ def test_arm_qnx_static(self):
 
     def test_arm_qnx_sqrt(self):
         ql = Qiling(["../examples/rootfs/arm_qnx/bin/hello_sqrt"], "../examples/rootfs/arm_qnx", verbose=QL_VERBOSE.DEBUG)
-        ql.run()        
+        ql.run()
+
+    def test_set_api_arm_qnx_sqrt(self):
+        self.set_api_puts_onenter = False
+        self.set_api_puts_onexit = False
+        self.set_api_printf_onenter = False
+        self.set_api_printf_onexit = False
+
+        def my_puts_onenter(ql: Qiling):
+            params = ql.os.resolve_fcall_params({'s': STRING})
+
+            print(f'puts("{params["s"]}")')
+            self.set_api_puts_onenter = True
+            return QL_CALL_BLOCK
+
+        def my_puts_onexit(ql: Qiling):
+            print(f'after puts')
+            self.set_api_puts_onexit = True
+            return QL_CALL_BLOCK
+
+        def my_printf_onenter(ql: Qiling):
+            params = ql.os.resolve_fcall_params({'s': STRING})
+
+            print(f'printf("{params["s"]}")')
+            self.set_api_printf_onenter = True
+            return QL_CALL_BLOCK
+
+        def my_printf_onexit(ql: Qiling):
+            print(f'after printf')
+            self.set_api_printf_onexit = True
+            return QL_CALL_BLOCK
+
+        ql = Qiling(["../examples/rootfs/arm_qnx/bin/hello_sqrt"], "../examples/rootfs/arm_qnx", verbose=QL_VERBOSE.DEBUG)
+        ql.set_api('puts', my_puts_onenter, QL_INTERCEPT.ENTER)
+        ql.set_api('printf', my_printf_onenter, QL_INTERCEPT.ENTER)
+        
+        # ql.set_api('puts', my_puts_onexit, QL_INTERCEPT.EXIT)
+        ql.set_api('printf', my_printf_onexit, QL_INTERCEPT.EXIT)
+
+        ql.run()
+
+        self.assertEqual(False, self.set_api_puts_onenter)
+        self.assertEqual(False, self.set_api_puts_onexit)
+        self.assertEqual(True, self.set_api_printf_onenter)
+        self.assertEqual(True, self.set_api_printf_onexit)
+
+        del self.set_api_puts_onenter
+        del self.set_api_puts_onexit
+        del self.set_api_printf_onenter
+        del self.set_api_printf_onexit
 
 if __name__ == "__main__":
     unittest.main()