From cb875f84889ad0883796e634d2cf4c2831d5ad3a Mon Sep 17 00:00:00 2001 From: xwings Date: Sun, 12 Apr 2026 16:48:53 +0800 Subject: [PATCH] more qiling style. update logging and exception --- qiling/exception.py | 3 ++ qiling/os/posix/kernel_proxy/__init__.py | 36 +++++++++++------------- qiling/os/posix/kernel_proxy/ipc.py | 6 ++-- qiling/os/posix/kernel_proxy/proxy.py | 7 +++-- tests/test_kernel_proxy.py | 5 ++-- 5 files changed, 31 insertions(+), 26 deletions(-) diff --git a/qiling/exception.py b/qiling/exception.py index e6d46060..a544b254 100644 --- a/qiling/exception.py +++ b/qiling/exception.py @@ -74,6 +74,9 @@ class QlMemoryMappedError(QlErrorBase): class QlGDTError(QlErrorBase): pass +class QlProxyConnectionError(QlErrorBase): + pass + class QlSyscallError(QlErrorBase): def __init__(self, errno, msg): super(QlSyscallError, self).__init__(msg) diff --git a/qiling/os/posix/kernel_proxy/__init__.py b/qiling/os/posix/kernel_proxy/__init__.py index b0fae205..f7ac75ea 100644 --- a/qiling/os/posix/kernel_proxy/__init__.py +++ b/qiling/os/posix/kernel_proxy/__init__.py @@ -24,18 +24,16 @@ import os import sys import socket import subprocess -import logging -from typing import Optional, TYPE_CHECKING +from typing import Dict, Optional, TYPE_CHECKING from qiling.const import QL_INTERCEPT, QL_OS +from qiling.exception import QlErrorArch, QlErrorSyscallError, QlErrorSyscallNotFound from qiling.os.posix.kernel_proxy.ipc import ProxyClient from qiling.os.posix.kernel_proxy.proxy_fd import ql_proxy_fd if TYPE_CHECKING: from qiling import Qiling -log = logging.getLogger(__name__) - class KernelProxy: """Forward specific syscalls to a real Linux kernel via a helper process. @@ -46,13 +44,13 @@ class KernelProxy: def __init__(self, ql: Qiling): if sys.platform != 'linux': - raise RuntimeError("KernelProxy requires a Linux host") + raise QlErrorArch("KernelProxy requires a Linux host") self.ql = ql self._process: Optional[subprocess.Popen] = None self._client: Optional[ProxyClient] = None - self._forwarded: dict = {} # name -> syscall_nr - self._reverse_table: Optional[dict] = None # name -> nr (built on first use) + self._forwarded: Dict[str, int] = {} # name -> syscall_nr + self._reverse_table: Optional[Dict[str, int]] = None # name -> nr (built on first use) self._start_proxy() @@ -78,14 +76,13 @@ class KernelProxy: child_sock.close() self._client = ProxyClient(parent_sock) - log.info(f"kernel proxy started (pid={self._process.pid})") + self.ql.log.info(f"kernel proxy started (pid={self._process.pid})") - def _build_reverse_table(self) -> dict: + def _build_reverse_table(self) -> Dict[str, int]: """Build name -> syscall_nr mapping from the guest architecture's syscall table.""" if self._reverse_table is not None: return self._reverse_table - from qiling.os.linux.map_syscall import get_syscall_mapper from qiling.const import QL_ARCH # get the raw syscall table dict for this architecture @@ -102,7 +99,7 @@ class KernelProxy: table_name = arch_tables.get(self.ql.arch.type) if table_name is None: - raise RuntimeError(f"KernelProxy: unsupported architecture {self.ql.arch.type}") + raise QlErrorArch(f"KernelProxy: unsupported architecture {self.ql.arch.type}") import qiling.os.linux.map_syscall as mod table = getattr(mod, table_name) @@ -115,7 +112,7 @@ class KernelProxy: """Resolve a syscall name to its number for the guest architecture.""" table = self._build_reverse_table() if name not in table: - raise ValueError( + raise QlErrorSyscallNotFound( f"KernelProxy: syscall '{name}' not found in {self.ql.arch.type.name} syscall table" ) return table[name] @@ -135,8 +132,8 @@ class KernelProxy: forwarder = self._make_forwarder(name, nr, returns_fd) self.ql.os.set_syscall(name, forwarder, QL_INTERCEPT.CALL) - log.info(f"forwarding syscall '{name}' (nr={nr}) to kernel proxy" - f"{' [returns FD]' if returns_fd else ''}") + self.ql.log.info(f"forwarding syscall '{name}' (nr={nr}) to kernel proxy" + f"{' [returns FD]' if returns_fd else ''}") def _make_forwarder(self, name: str, guest_nr: int, returns_fd: bool): """Create a CALL hook closure for one syscall.""" @@ -170,17 +167,16 @@ class KernelProxy: self._host_table = self._load_host_syscall_table() if name not in self._host_table: - raise RuntimeError(f"KernelProxy: syscall '{name}' not available on host") + raise QlErrorSyscallNotFound(f"KernelProxy: syscall '{name}' not available on host") return self._host_table[name] - def _load_host_syscall_table(self) -> dict: + def _load_host_syscall_table(self) -> Dict[str, int]: """Load the host's syscall name->nr mapping. Uses the same Qiling tables, indexed by the host architecture. """ import platform - from qiling.const import QL_ARCH import qiling.os.linux.map_syscall as mod machine = platform.machine() @@ -195,7 +191,7 @@ class KernelProxy: table_name = host_arch_map.get(machine) if table_name is None: - raise RuntimeError(f"KernelProxy: unsupported host architecture '{machine}'") + raise QlErrorArch(f"KernelProxy: unsupported host architecture '{machine}'") table = getattr(mod, table_name) return {name: nr for nr, name in table.items()} @@ -208,7 +204,7 @@ class KernelProxy: ql.os.fd[i] = fd_obj return i - raise OSError("kernel_proxy: FD table full") + raise QlErrorSyscallError("kernel_proxy: FD table full") def stop(self): """Stop the proxy process.""" @@ -226,7 +222,7 @@ class KernelProxy: except subprocess.TimeoutExpired: self._process.kill() self._process.wait() - log.info(f"kernel proxy stopped (pid={self._process.pid})") + self.ql.log.info(f"kernel proxy stopped (pid={self._process.pid})") self._process = None def __del__(self): diff --git a/qiling/os/posix/kernel_proxy/ipc.py b/qiling/os/posix/kernel_proxy/ipc.py index 8bf26eaa..59202c19 100644 --- a/qiling/os/posix/kernel_proxy/ipc.py +++ b/qiling/os/posix/kernel_proxy/ipc.py @@ -17,6 +17,8 @@ import struct import socket from enum import IntEnum +from qiling.exception import QlProxyConnectionError + class MsgType(IntEnum): SYSCALL = 1 @@ -64,7 +66,7 @@ def _recvall(sock: socket.socket, n: int) -> bytes: while len(buf) < n: chunk = sock.recv(n - len(buf)) if not chunk: - raise ConnectionError("kernel proxy connection closed") + raise QlProxyConnectionError("kernel proxy connection closed") buf.extend(chunk) return bytes(buf) @@ -162,7 +164,7 @@ class ProxyServer: return MsgType.FD_OP, (FdOp(op), proxy_fd, arg1, arg2, data) else: - raise ValueError(f"unknown message type: {msg_type}") + raise QlProxyConnectionError(f"unknown message type: {msg_type}") def send_syscall_response(self, retval: int, errno_val: int): payload = struct.pack(SYSCALL_RESP_FMT, retval, errno_val) diff --git a/qiling/os/posix/kernel_proxy/proxy.py b/qiling/os/posix/kernel_proxy/proxy.py index 41bceac0..af4934bd 100644 --- a/qiling/os/posix/kernel_proxy/proxy.py +++ b/qiling/os/posix/kernel_proxy/proxy.py @@ -13,6 +13,7 @@ Usage (internal — started by KernelProxy.__init__): python -m qiling.os.posix.kernel_proxy.proxy """ +import logging import os import sys import ctypes @@ -24,10 +25,12 @@ from qiling.os.posix.kernel_proxy.ipc import ( ProxyServer, MsgType, FdOp ) +log = logging.getLogger("qiling.os.posix.kernel_proxy.proxy") + # load libc for raw syscall() _libc_path = ctypes.util.find_library("c") if _libc_path is None: - print("kernel_proxy: cannot find libc", file=sys.stderr) + log.critical("kernel_proxy: cannot find libc") sys.exit(1) _libc = ctypes.CDLL(_libc_path, use_errno=True) @@ -86,7 +89,7 @@ def handle_fd_op(op: FdOp, proxy_fd: int, arg1: int, arg2: int, data: bytes) -> def main(): if len(sys.argv) != 2: - print(f"usage: {sys.argv[0]} ", file=sys.stderr) + log.error(f"usage: {sys.argv[0]} ") sys.exit(1) sock_fd = int(sys.argv[1]) diff --git a/tests/test_kernel_proxy.py b/tests/test_kernel_proxy.py index d6690ad7..52495238 100644 --- a/tests/test_kernel_proxy.py +++ b/tests/test_kernel_proxy.py @@ -283,13 +283,14 @@ class KernelProxyTest(unittest.TestCase): # ------------------------------------------------------------------------- def test_forward_invalid_syscall_name(self): - """forward_syscall with bogus name raises ValueError.""" + """forward_syscall with bogus name raises QlErrorSyscallNotFound.""" from qiling.os.posix.kernel_proxy import KernelProxy + from qiling.exception import QlErrorSyscallNotFound ql = Qiling([self.HELLO_BIN], self.ROOTFS, verbose=QL_VERBOSE.OFF) proxy = KernelProxy(ql) - with self.assertRaises(ValueError): + with self.assertRaises(QlErrorSyscallNotFound): proxy.forward_syscall('nonexistent_syscall_xyz') proxy.stop()