better New SYSCALL_EX IPC message, FD arg type and Forwarder closure

qiling/os/posix/kernel_proxy/__init__.py

@@ -8,13 +8,22 @@ Hybrid kernel proxy — forward specific syscalls to a real Linux kernel.
 
 Usage:
     from qiling import Qiling
-    from qiling.os.posix.kernel_proxy import KernelProxy
+    from qiling.os.posix.kernel_proxy import KernelProxy, FD, PtrIn, PtrOut
 
     ql = Qiling(argv=["/bin/myserver"], rootfs="rootfs/x8664_linux")
     proxy = KernelProxy(ql)
-    proxy.forward_syscall("epoll_create", returns_fd=True)
-    proxy.forward_syscall("epoll_ctl")
-    proxy.forward_syscall("epoll_wait")
+
+    # integer-arg syscall returning a new FD
+    proxy.forward_syscall("epoll_create1", returns_fd=True)
+
+    # epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)
+    proxy.forward_syscall("epoll_ctl",
+                          arg_types=(FD, "int", FD, PtrIn(size=12)))
+
+    # epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout)
+    proxy.forward_syscall("epoll_wait",
+                          arg_types=(FD, PtrOut(size=lambda a: a[2] * 12), "int", "int"))
+
     ql.run()
 """
 
@@ -24,10 +33,14 @@ import os
 import sys
 import socket
 import subprocess
-from typing import Dict, Optional, TYPE_CHECKING
+import weakref
+from typing import Dict, Optional, Sequence, Tuple, TYPE_CHECKING
 
-from qiling.const import QL_INTERCEPT, QL_OS
+from qiling.const import QL_INTERCEPT
 from qiling.exception import QlErrorArch, QlErrorSyscallError, QlErrorSyscallNotFound
+from qiling.os.posix.kernel_proxy.argtypes import (
+    INT, FD, PtrIn, PtrOut, PtrInOut, is_pointer,
+)
 from qiling.os.posix.kernel_proxy.ipc import ProxyClient
 from qiling.os.posix.kernel_proxy.proxy_fd import ql_proxy_fd
 
@@ -35,6 +48,9 @@ if TYPE_CHECKING:
     from qiling import Qiling
 
 
+__all__ = ['KernelProxy', 'INT', 'FD', 'PtrIn', 'PtrOut', 'PtrInOut']
+
+
 class KernelProxy:
     """Forward specific syscalls to a real Linux kernel via a helper process.
 
@@ -117,40 +133,74 @@ class KernelProxy:
             )
         return table[name]
 
-    def forward_syscall(self, name: str, returns_fd: bool = False):
+    def forward_syscall(self, name: str, returns_fd: bool = False,
+                        arg_types: Optional[Sequence] = None):
         """Register a CALL hook that forwards this syscall to the kernel proxy.
 
         Args:
-            name: syscall name (e.g. "epoll_create", "eventfd2")
+            name: syscall name (e.g. "epoll_create1", "eventfd2").
             returns_fd: if True, wrap the return value in ql_proxy_fd and store
-                        in the Qiling FD table. Use this for syscalls that return
-                        file descriptors (epoll_create, eventfd, timerfd_create, etc.)
+                        it in the Qiling FD table. Use this for syscalls that
+                        return file descriptors (epoll_create1, eventfd2, etc.).
+            arg_types: optional per-arg descriptors. Each entry is one of:
+                       INT (or "int") — pass through unchanged (default).
+                       FD (or "fd")   — guest FD; translated to the proxy FD.
+                       PtrIn(size)    — pointer; bytes copied from guest to proxy.
+                       PtrOut(size)   — pointer; bytes copied back from proxy to guest.
+                       PtrInOut(size) — pointer; both directions.
+                       If omitted, all arguments are treated as INT.
         """
         nr = self._resolve_syscall_nr(name)
         self._forwarded[name] = nr
 
-        forwarder = self._make_forwarder(name, nr, returns_fd)
+        forwarder = self._make_forwarder(name, nr, returns_fd, arg_types)
         self.ql.os.set_syscall(name, forwarder, QL_INTERCEPT.CALL)
 
-        self.ql.log.info(f"forwarding syscall '{name}' (nr={nr}) to kernel proxy"
-                         f"{' [returns FD]' if returns_fd else ''}")
+        kind = []
+        if returns_fd:
+            kind.append('returns FD')
+        if arg_types:
+            kind.append(f'arg_types={tuple(type(a).__name__ if not isinstance(a, str) else a for a in arg_types)}')
 
-    def _make_forwarder(self, name: str, guest_nr: int, returns_fd: bool):
-        """Create a CALL hook closure for one syscall."""
+        suffix = f" [{', '.join(kind)}]" if kind else ''
+        self.ql.log.info(f"forwarding syscall '{name}' (nr={nr}) to kernel proxy{suffix}")
+
+    def _make_forwarder(self, name: str, guest_nr: int, returns_fd: bool,
+                        arg_types: Optional[Sequence]):
+        """Create a CALL hook closure for one syscall.
+
+        Captures only the data the closure needs (host syscall nr, client, weakref
+        to self) so the registered hook does not keep the KernelProxy alive.
+        """
+        # resolve once at registration time so the hot path stays simple
+        host_nr = self._get_host_syscall_nr(name)
         client = self._client
+        weak_self = weakref.ref(self)
+
+        # normalize arg_types to a tuple, treating the string aliases as-is
+        spec = tuple(arg_types) if arg_types else ()
+        has_pointers = any(is_pointer(s) for s in spec)
 
         def _forwarder(ql, *args):
-            # use the HOST syscall number, not the guest number.
-            # for now, resolve from the host's syscall table at runtime.
-            host_nr = self._get_host_syscall_nr(name)
+            self_ref = weak_self()
+            if self_ref is None:
+                ql.log.error(f"kernel_proxy: {name}() called after proxy was destroyed")
+                return -1
 
-            padded = args + (0,) * (6 - len(args))
-            retval = client.syscall(host_nr, padded[:6])
+            translated = self_ref._translate_args(name, args, spec)
+
+            if has_pointers:
+                in_bufs, out_specs, out_arg_indices = self_ref._collect_buffers(
+                    ql, translated, spec
+                )
+                retval, out_data = client.syscall_ex(host_nr, translated, in_bufs, out_specs)
+                self_ref._writeback_buffers(ql, args, out_arg_indices, out_data)
+            else:
+                retval = client.syscall(host_nr, translated)
 
             if returns_fd and retval >= 0:
-                # the proxy created a real FD. wrap it and store in Qiling's FD table.
                 proxy_fd_obj = ql_proxy_fd(client, retval)
-                guest_fd = self._alloc_fd(ql, proxy_fd_obj)
+                guest_fd = self_ref._alloc_fd(ql, proxy_fd_obj)
                 ql.log.debug(f"kernel_proxy: {name}() -> proxy_fd={retval}, guest_fd={guest_fd}")
                 return guest_fd
 
@@ -160,6 +210,60 @@ class KernelProxy:
         _forwarder.__name__ = f'ql_syscall_{name}'
         return _forwarder
 
+    def _translate_args(self, name: str, args: Tuple[int, ...],
+                        spec: Tuple) -> Tuple[int, ...]:
+        """Translate guest FD args to proxy FD numbers; pad to 6 args.
+
+        Pointer args are left untouched here — _collect_buffers replaces them
+        with the proxy-side buffer addresses just before invocation.
+        """
+        out = list(args) + [0] * (6 - len(args))
+
+        for idx, kind in enumerate(spec):
+            if kind == FD:
+                guest_fd = args[idx]
+                fd_obj = self.ql.os.fd[guest_fd] if 0 <= guest_fd < len(self.ql.os.fd) else None
+
+                if not isinstance(fd_obj, ql_proxy_fd):
+                    raise QlErrorSyscallError(
+                        f"kernel_proxy: {name}() arg{idx} guest_fd={guest_fd} "
+                        f"does not refer to a proxy-owned FD"
+                    )
+
+                out[idx] = fd_obj._proxy_fd
+
+        return tuple(out[:6])
+
+    def _collect_buffers(self, ql, args: Tuple[int, ...], spec: Tuple):
+        """Read PtrIn/PtrInOut buffers from guest memory; collect PtrOut sizes."""
+        in_bufs = []
+        out_specs = []
+        out_arg_indices = []
+
+        for idx, kind in enumerate(spec):
+            if isinstance(kind, (PtrIn, PtrInOut)):
+                size = kind.resolve(args)
+                if size > 0:
+                    data = bytes(ql.mem.read(args[idx], size))
+                    in_bufs.append((idx, data))
+
+            if isinstance(kind, (PtrOut, PtrInOut)):
+                size = kind.resolve(args)
+                if size > 0:
+                    out_specs.append((idx, size))
+                    out_arg_indices.append(idx)
+
+        return in_bufs, out_specs, out_arg_indices
+
+    @staticmethod
+    def _writeback_buffers(ql, args: Tuple[int, ...],
+                           out_arg_indices: Sequence[int],
+                           out_data: Sequence[bytes]):
+        """Write PtrOut/PtrInOut response buffers back into guest memory."""
+        for idx, data in zip(out_arg_indices, out_data):
+            if data:
+                ql.mem.write(args[idx], data)
+
     def _get_host_syscall_nr(self, name: str) -> int:
         """Get the syscall number on the HOST architecture."""
         # we are running on Linux — read from the host's syscall table

qiling/os/posix/kernel_proxy/argtypes.py

@@ -0,0 +1,73 @@
+#!/usr/bin/env python3
+#
+# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
+#
+
+"""
+Argument type descriptors for forwarded syscalls.
+
+When forwarding a syscall whose arguments are not all plain integers, the user
+declares each argument's role so the forwarder knows how to marshal it:
+
+    INT       — pass through unchanged (the default).
+    FD        — guest file descriptor; translate to the proxy-side FD before
+                forwarding. If the FD does not refer to a ql_proxy_fd, an error
+                is raised.
+    PtrIn(s)  — pointer to a buffer of `s` bytes. Read from guest memory and
+                copied to the proxy.
+    PtrOut(s) — pointer to a buffer of `s` bytes. Allocated on the proxy and
+                copied back to guest memory after the syscall.
+    PtrInOut  — both directions.
+
+`s` may be an integer or a callable taking the raw arg tuple and returning the
+buffer length (e.g. for syscalls where the size depends on another argument).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Callable, Tuple, Union
+
+
+INT = 'int'
+FD = 'fd'
+
+SizeSpec = Union[int, Callable[[Tuple[int, ...]], int]]
+
+
+def _resolve_size(size: SizeSpec, args: Tuple[int, ...]) -> int:
+    if callable(size):
+        return int(size(args))
+
+    return int(size)
+
+
+@dataclass(frozen=True)
+class PtrIn:
+    """Input pointer — buffer of `size` bytes is read from guest memory."""
+    size: SizeSpec
+
+    def resolve(self, args: Tuple[int, ...]) -> int:
+        return _resolve_size(self.size, args)
+
+
+@dataclass(frozen=True)
+class PtrOut:
+    """Output pointer — buffer of `size` bytes is written to guest memory."""
+    size: SizeSpec
+
+    def resolve(self, args: Tuple[int, ...]) -> int:
+        return _resolve_size(self.size, args)
+
+
+@dataclass(frozen=True)
+class PtrInOut:
+    """In/out pointer — buffer is read from guest, then written back."""
+    size: SizeSpec
+
+    def resolve(self, args: Tuple[int, ...]) -> int:
+        return _resolve_size(self.size, args)
+
+
+def is_pointer(spec) -> bool:
+    return isinstance(spec, (PtrIn, PtrOut, PtrInOut))

qiling/os/posix/kernel_proxy/ipc.py

@@ -6,9 +6,10 @@
 """
 IPC protocol between Qiling and the kernel proxy process.
 
-Two message types:
-  SYSCALL  — forward a raw syscall (number + 6 integer args)
-  FD_OP    — perform an operation on a proxy-side FD (read/write/close/dup/fcntl/ioctl)
+Three message types:
+  SYSCALL    — forward a raw syscall (number + 6 integer args, no buffers)
+  SYSCALL_EX — forward a syscall with input/output buffer marshaling
+  FD_OP      — perform an operation on a proxy-side FD (read/write/close/dup/fcntl/ioctl)
 
 All messages are length-prefixed binary over a Unix socketpair.
 """
@@ -16,13 +17,15 @@ All messages are length-prefixed binary over a Unix socketpair.
 import struct
 import socket
 from enum import IntEnum
+from typing import List, Sequence, Tuple
 
 from qiling.exception import QlProxyConnectionError
 
 
 class MsgType(IntEnum):
-    SYSCALL = 1
-    FD_OP   = 2
+    SYSCALL    = 1
+    FD_OP      = 2
+    SYSCALL_EX = 3
 
 
 class FdOp(IntEnum):
@@ -35,14 +38,19 @@ class FdOp(IntEnum):
 
 
 # Wire format:
-#   Request header:  [msg_type: u8][payload_len: u32]
-#   SYSCALL payload: [syscall_nr: u32][args: 6 x i64]
-#   FD_OP payload:   [op: u8][proxy_fd: i32][arg1: i64][arg2: i64][data_len: u32][data: bytes]
+#   Request header:     [msg_type: u8][payload_len: u32]
+#   SYSCALL payload:    [syscall_nr: u32][args: 6 x i64]
+#   SYSCALL_EX payload: [syscall_nr: u32][args: 6 x i64]
+#                       [num_in: u8] then num_in * [arg_idx: u8][len: u32][data: bytes]
+#                       [num_out: u8] then num_out * [arg_idx: u8][len: u32]
+#   FD_OP payload:      [op: u8][proxy_fd: i32][arg1: i64][arg2: i64][data_len: u32][data: bytes]
 #
 #   Response header: [status: i8][payload_len: u32]
 #   status 0 = success, -1 = error
-#   SYSCALL response payload: [return_value: i64][errno: i32]
-#   FD_OP response payload:   [return_value: i64][errno: i32][data_len: u32][data: bytes]
+#   SYSCALL response payload:    [return_value: i64][errno: i32]
+#   SYSCALL_EX response payload: [return_value: i64][errno: i32][num_out: u8]
+#                                then num_out * [len: u32][data: bytes]
+#   FD_OP response payload:      [return_value: i64][errno: i32][data_len: u32][data: bytes]
 
 HEADER_FMT = '!BI'           # msg_type/status (u8) + payload_len (u32)
 HEADER_SIZE = struct.calcsize(HEADER_FMT)
@@ -53,6 +61,9 @@ SYSCALL_REQ_SIZE = struct.calcsize(SYSCALL_REQ_FMT)
 SYSCALL_RESP_FMT = '!qi'     # return_value (i64) + errno (i32)
 SYSCALL_RESP_SIZE = struct.calcsize(SYSCALL_RESP_FMT)
 
+SYSCALL_EX_RESP_HEAD_FMT = '!qiB'   # return_value (i64) + errno (i32) + num_out (u8)
+SYSCALL_EX_RESP_HEAD_SIZE = struct.calcsize(SYSCALL_EX_RESP_HEAD_FMT)
+
 FD_OP_REQ_FMT = '!BiqqI'    # op (u8) + proxy_fd (i32) + arg1 (i64) + arg2 (i64) + data_len (u32)
 FD_OP_REQ_SIZE = struct.calcsize(FD_OP_REQ_FMT)
 
@@ -94,6 +105,54 @@ class ProxyClient:
         retval, errno_val = struct.unpack(SYSCALL_RESP_FMT, resp_payload)
         return retval
 
+    def syscall_ex(self, nr: int, args: Sequence[int],
+                   in_bufs: Sequence[Tuple[int, bytes]],
+                   out_specs: Sequence[Tuple[int, int]]) -> Tuple[int, List[bytes]]:
+        """Forward a syscall with buffer marshaling.
+
+        Args:
+            nr: host syscall number.
+            args: 6 integer arg values; for buffer args these are placeholders —
+                  the proxy replaces them with the buffer address before invoking.
+            in_bufs: list of (arg_idx, data) — buffers to copy in.
+            out_specs: list of (arg_idx, length) — buffers to copy out.
+
+        Returns:
+            (retval, out_bufs) where out_bufs is a list aligned with out_specs.
+        """
+        padded = tuple(args) + (0,) * (6 - len(args))
+        payload = bytearray(struct.pack(SYSCALL_REQ_FMT, nr, *padded[:6]))
+
+        payload.append(len(in_bufs))
+        for arg_idx, data in in_bufs:
+            payload += struct.pack('!BI', arg_idx, len(data))
+            payload += data
+
+        payload.append(len(out_specs))
+        for arg_idx, length in out_specs:
+            payload += struct.pack('!BI', arg_idx, length)
+
+        header = struct.pack(HEADER_FMT, MsgType.SYSCALL_EX, len(payload))
+        self._sock.sendall(header + bytes(payload))
+
+        resp_header = _recvall(self._sock, HEADER_SIZE)
+        _, resp_len = struct.unpack(HEADER_FMT, resp_header)
+        resp_payload = _recvall(self._sock, resp_len)
+
+        retval, _errno, num_out = struct.unpack(
+            SYSCALL_EX_RESP_HEAD_FMT, resp_payload[:SYSCALL_EX_RESP_HEAD_SIZE]
+        )
+
+        out_bufs: List[bytes] = []
+        offset = SYSCALL_EX_RESP_HEAD_SIZE
+        for _ in range(num_out):
+            (length,) = struct.unpack('!I', resp_payload[offset:offset + 4])
+            offset += 4
+            out_bufs.append(resp_payload[offset:offset + length])
+            offset += length
+
+        return retval, out_bufs
+
     def _fd_op(self, op: FdOp, proxy_fd: int, arg1: int = 0, arg2: int = 0, data: bytes = b'') -> tuple:
         """Send an FD operation. Returns (return_value, data)."""
         payload = struct.pack(FD_OP_REQ_FMT, op, proxy_fd, arg1, arg2, len(data))
@@ -157,6 +216,31 @@ class ProxyServer:
             fields = struct.unpack(SYSCALL_REQ_FMT, payload)
             return MsgType.SYSCALL, fields  # (nr, a0, a1, a2, a3, a4, a5)
 
+        elif msg_type == MsgType.SYSCALL_EX:
+            offset = SYSCALL_REQ_SIZE
+            fixed = struct.unpack(SYSCALL_REQ_FMT, payload[:offset])
+            nr = fixed[0]
+            args = list(fixed[1:])
+
+            num_in = payload[offset]
+            offset += 1
+            in_bufs: List[Tuple[int, bytes]] = []
+            for _ in range(num_in):
+                arg_idx, length = struct.unpack('!BI', payload[offset:offset + 5])
+                offset += 5
+                in_bufs.append((arg_idx, payload[offset:offset + length]))
+                offset += length
+
+            num_out = payload[offset]
+            offset += 1
+            out_specs: List[Tuple[int, int]] = []
+            for _ in range(num_out):
+                arg_idx, length = struct.unpack('!BI', payload[offset:offset + 5])
+                offset += 5
+                out_specs.append((arg_idx, length))
+
+            return MsgType.SYSCALL_EX, (nr, args, in_bufs, out_specs)
+
         elif msg_type == MsgType.FD_OP:
             fixed = struct.unpack(FD_OP_REQ_FMT, payload[:FD_OP_REQ_SIZE])
             op, proxy_fd, arg1, arg2, data_len = fixed
@@ -171,6 +255,15 @@ class ProxyServer:
         header = struct.pack(HEADER_FMT, 0, len(payload))
         self._sock.sendall(header + payload)
 
+    def send_syscall_ex_response(self, retval: int, errno_val: int, out_bufs: Sequence[bytes]):
+        payload = bytearray(struct.pack(SYSCALL_EX_RESP_HEAD_FMT, retval, errno_val, len(out_bufs)))
+        for buf in out_bufs:
+            payload += struct.pack('!I', len(buf))
+            payload += buf
+
+        header = struct.pack(HEADER_FMT, 0, len(payload))
+        self._sock.sendall(header + bytes(payload))
+
     def send_fd_op_response(self, retval: int, errno_val: int, data: bytes = b''):
         payload = struct.pack(FD_OP_RESP_FMT, retval, errno_val, len(data))
         payload += data

qiling/os/posix/kernel_proxy/proxy.py

@@ -51,6 +51,36 @@ def raw_syscall(nr: int, a0: int, a1: int, a2: int, a3: int, a4: int, a5: int) -
     return result, 0
 
 
+def raw_syscall_ex(nr: int, args: list, in_bufs: list, out_specs: list) -> tuple:
+    """Execute a syscall with buffer marshaling.
+
+    For each in_buf (arg_idx, data): allocate a ctypes buffer initialized with
+    data and place its address in args[arg_idx]. For each out_spec (arg_idx, length):
+    allocate a zeroed buffer and place its address in args[arg_idx]. After the
+    syscall, return the contents of each out buffer.
+    """
+    keepalive = []  # keep ctypes buffers alive until after the syscall
+    out_buffers = []  # parallel to out_specs
+
+    args = list(args)
+
+    for arg_idx, data in in_bufs:
+        buf = ctypes.create_string_buffer(data, len(data))
+        keepalive.append(buf)
+        args[arg_idx] = ctypes.addressof(buf)
+
+    for arg_idx, length in out_specs:
+        buf = ctypes.create_string_buffer(length)
+        keepalive.append(buf)
+        out_buffers.append(buf)
+        args[arg_idx] = ctypes.addressof(buf)
+
+    retval, err = raw_syscall(nr, *args)
+
+    out_data = [bytes(buf.raw) for buf in out_buffers]
+    return retval, err, out_data
+
+
 def handle_fd_op(op: FdOp, proxy_fd: int, arg1: int, arg2: int, data: bytes) -> tuple:
     """Handle an FD operation on a proxy-side FD. Returns (retval, errno, data)."""
     try:
@@ -107,6 +137,11 @@ def main():
             retval, err = raw_syscall(nr, a0, a1, a2, a3, a4, a5)
             server.send_syscall_response(retval, err)
 
+        elif msg_type == MsgType.SYSCALL_EX:
+            nr, args, in_bufs, out_specs = fields
+            retval, err, out_bufs = raw_syscall_ex(nr, args, in_bufs, out_specs)
+            server.send_syscall_ex_response(retval, err, out_bufs)
+
         elif msg_type == MsgType.FD_OP:
             op, proxy_fd, arg1, arg2, data = fields
             retval, err, resp_data = handle_fd_op(op, proxy_fd, arg1, arg2, data)

tests/test_kernel_proxy.py

@@ -352,5 +352,173 @@ class KernelProxyTest(unittest.TestCase):
         del ql
 
 
+    # -------------------------------------------------------------------------
+    # FD translation (#3)
+    # -------------------------------------------------------------------------
+
+    def test_fd_arg_translated_to_proxy_fd(self):
+        """When a forwarded syscall's arg is declared FD, the guest fd is replaced
+        with the underlying proxy fd before forwarding."""
+        from qiling.os.posix.kernel_proxy import KernelProxy, FD
+        from qiling.os.posix.kernel_proxy.proxy_fd import ql_proxy_fd
+
+        ql = Qiling([self.HELLO_BIN], self.ROOTFS, verbose=QL_VERBOSE.OFF)
+        proxy = KernelProxy(ql)
+        proxy.forward_syscall('close', arg_types=(FD,))
+
+        # capture what gets sent over IPC
+        sent = []
+        original_syscall = proxy._client.syscall
+
+        def spy(nr, args):
+            sent.append((nr, tuple(args)))
+            return 0  # pretend the kernel accepted
+
+        proxy._client.syscall = spy
+
+        # plant a ql_proxy_fd at guest_fd=42 with a chosen proxy_fd value
+        guest_fd = 42
+        fake_proxy_fd = 9999
+        ql.os.fd[guest_fd] = ql_proxy_fd(proxy._client, fake_proxy_fd)
+
+        close_hook = ql.os.posix_syscall_hooks[QL_INTERCEPT.CALL]['ql_syscall_close']
+        close_hook(ql, guest_fd)
+
+        self.assertEqual(len(sent), 1)
+        _nr, args = sent[0]
+        self.assertEqual(args[0], fake_proxy_fd,
+                         f"expected proxy_fd={fake_proxy_fd} forwarded, got args={args}")
+
+        proxy._client.syscall = original_syscall
+        ql.os.fd[guest_fd] = None
+        proxy.stop()
+        del ql
+
+    def test_fd_arg_rejects_non_proxy_fd(self):
+        """Forwarding with FD arg type rejects non-proxy guest FDs."""
+        from qiling.os.posix.kernel_proxy import KernelProxy, FD
+        from qiling.exception import QlErrorSyscallError
+
+        ql = Qiling([self.HELLO_BIN], self.ROOTFS, verbose=QL_VERBOSE.OFF)
+        proxy = KernelProxy(ql)
+        proxy.forward_syscall('close', arg_types=(FD,))
+
+        close_hook = ql.os.posix_syscall_hooks[QL_INTERCEPT.CALL]['ql_syscall_close']
+
+        # stdin (fd 0) is a regular ql_pipe, not a proxy fd
+        with self.assertRaises(QlErrorSyscallError):
+            close_hook(ql, 0)
+
+        proxy.stop()
+        del ql
+
+    # -------------------------------------------------------------------------
+    # Pointer marshaling (#2)
+    # -------------------------------------------------------------------------
+
+    def test_ptr_out_writes_back_to_guest_memory(self):
+        """PtrOut buffer is written back into guest memory after the syscall.
+
+        Uses pipe2(int pipefd[2], int flags) — pipefd is an output buffer of
+        2 * sizeof(int) = 8 bytes containing the read/write FDs created by
+        the kernel.
+        """
+        from qiling.os.posix.kernel_proxy import KernelProxy, PtrOut
+
+        ql = Qiling([self.HELLO_BIN], self.ROOTFS, verbose=QL_VERBOSE.OFF)
+        proxy = KernelProxy(ql)
+        proxy.forward_syscall('pipe2', arg_types=(PtrOut(size=8), 'int'))
+
+        # pick a free guest address and map it
+        addr = 0x800000
+        ql.mem.map(addr, 0x1000)
+        ql.mem.write(addr, b'\xff' * 8)  # poison so we can detect the writeback
+
+        hook = ql.os.posix_syscall_hooks[QL_INTERCEPT.CALL]['ql_syscall_pipe2']
+        retval = hook(ql, addr, 0)
+        self.assertEqual(retval, 0)
+
+        raw = bytes(ql.mem.read(addr, 8))
+        rfd, wfd = struct.unpack('<ii', raw)
+        # the proxy returns kernel-side FDs; both must be valid (>=0) and distinct
+        self.assertGreaterEqual(rfd, 0)
+        self.assertGreaterEqual(wfd, 0)
+        self.assertNotEqual(rfd, wfd)
+
+        # clean up the proxy-side FDs the kernel just gave us
+        import os as _os
+        _os.close(rfd)
+        _os.close(wfd)
+
+        proxy.stop()
+        del ql
+
+    def test_ptr_in_reads_guest_memory(self):
+        """PtrIn buffer is copied from guest memory and the proxy sees the data.
+
+        Uses write(int fd, const void *buf, size_t count) on stderr (fd 2).
+        We can't easily inspect proxy's stderr, but a successful return value
+        equal to count proves the data was forwarded — write would otherwise
+        return -EFAULT for a bad pointer or short for less data.
+        """
+        from qiling.os.posix.kernel_proxy import KernelProxy, PtrIn
+
+        ql = Qiling([self.HELLO_BIN], self.ROOTFS, verbose=QL_VERBOSE.OFF)
+        proxy = KernelProxy(ql)
+        proxy.forward_syscall('write', arg_types=('int', PtrIn(size=lambda a: a[2]), 'int'))
+
+        addr = 0x800000
+        payload = b'kernel_proxy ptr_in roundtrip\n'
+        ql.mem.map(addr, 0x1000)
+        ql.mem.write(addr, payload)
+
+        # forward stderr (fd 2 in the proxy is our subprocess's stderr,
+        # which goes to the test runner — harmless)
+        hook = ql.os.posix_syscall_hooks[QL_INTERCEPT.CALL]['ql_syscall_write']
+        retval = hook(ql, 2, addr, len(payload))
+        self.assertEqual(retval, len(payload))
+
+        proxy.stop()
+        del ql
+
+    def test_ptr_size_callable(self):
+        """PtrIn/PtrOut accept a size callable that depends on other args."""
+        from qiling.os.posix.kernel_proxy import PtrIn, PtrOut
+
+        ptr = PtrIn(size=lambda args: args[2] * 4)
+        self.assertEqual(ptr.resolve((0, 0, 5)), 20)
+
+        ptr2 = PtrOut(size=12)
+        self.assertEqual(ptr2.resolve((0, 0, 0)), 12)
+
+    # -------------------------------------------------------------------------
+    # Reference cycle (#4)
+    # -------------------------------------------------------------------------
+
+    def test_no_reference_cycle_via_hook(self):
+        """The forwarder closure holds only a weakref to KernelProxy, so the
+        proxy can be garbage-collected once the user drops their reference,
+        even though the hook is still registered on ql.os."""
+        import gc
+        import weakref
+        from qiling.os.posix.kernel_proxy import KernelProxy
+
+        ql = Qiling([self.HELLO_BIN], self.ROOTFS, verbose=QL_VERBOSE.OFF)
+        proxy = KernelProxy(ql)
+        proxy.forward_syscall('getpid')
+        proxy.forward_syscall('eventfd2', returns_fd=True)
+
+        wref = weakref.ref(proxy)
+        proxy.stop()  # tear down the subprocess but leave the hooks registered
+        del proxy
+        gc.collect()
+
+        self.assertIsNone(wref(),
+                          "KernelProxy survived after stop()+del — closure must hold "
+                          "a strong ref (cycle), defeating the weakref design")
+
+        del ql
+
+
 if __name__ == "__main__":
     unittest.main()