more qiling style. update logging and exception

Author: xwings

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)

qiling/os/posix/kernel_proxy/__init__.py

@@ -24,18 +24,16 @@
 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 @@ def _start_proxy(self):
         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 @@ def _build_reverse_table(self) -> dict:
 
         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 @@ def _resolve_syscall_nr(self, name: str) -> int:
         """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 @@ def forward_syscall(self, name: str, returns_fd: bool = False):
         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 @@ def _get_host_syscall_nr(self, name: str) -> int:
             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 @@ def _load_host_syscall_table(self) -> dict:
 
         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 @@ def _alloc_fd(ql, fd_obj) -> int:
                 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 @@ def stop(self):
             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):

qiling/os/posix/kernel_proxy/ipc.py

@@ -17,6 +17,8 @@
 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 @@ def recv_request(self) -> tuple:
             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)

qiling/os/posix/kernel_proxy/proxy.py

@@ -13,6 +13,7 @@
     python -m qiling.os.posix.kernel_proxy.proxy <socket_fd>
 """
 
+import logging
 import os
 import sys
 import ctypes
@@ -24,10 +25,12 @@
     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]} <socket_fd>", file=sys.stderr)
+        log.error(f"usage: {sys.argv[0]} <socket_fd>")
         sys.exit(1)
 
     sock_fd = int(sys.argv[1])

tests/test_kernel_proxy.py

@@ -283,13 +283,14 @@ def test_ipc_roundtrip(self):
     # -------------------------------------------------------------------------
 
     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()