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,
(nr: int, args: list, in_bufs: list, out_specs: list)
| 52 | |
| 53 | |
| 54 | def raw_syscall_ex(nr: int, args: list, in_bufs: list, out_specs: list) -> tuple: |
| 55 | """Execute a syscall with buffer marshaling. |
| 56 | |
| 57 | For each in_buf (arg_idx, data): allocate a ctypes buffer initialized with |
| 58 | data and place its address in args[arg_idx]. For each out_spec (arg_idx, length): |
| 59 | allocate a zeroed buffer and place its address in args[arg_idx]. After the |
| 60 | syscall, return the contents of each out buffer. |
| 61 | """ |
| 62 | keepalive = [] # keep ctypes buffers alive until after the syscall |
| 63 | out_buffers = [] # parallel to out_specs |
| 64 | |
| 65 | args = list(args) |
| 66 | |
| 67 | for arg_idx, data in in_bufs: |
| 68 | buf = ctypes.create_string_buffer(data, len(data)) |
| 69 | keepalive.append(buf) |
| 70 | args[arg_idx] = ctypes.addressof(buf) |
| 71 | |
| 72 | for arg_idx, length in out_specs: |
| 73 | buf = ctypes.create_string_buffer(length) |
| 74 | keepalive.append(buf) |
| 75 | out_buffers.append(buf) |
| 76 | args[arg_idx] = ctypes.addressof(buf) |
| 77 | |
| 78 | retval, err = raw_syscall(nr, *args) |
| 79 | |
| 80 | out_data = [bytes(buf.raw) for buf in out_buffers] |
| 81 | return retval, err, out_data |
| 82 | |
| 83 | |
| 84 | def handle_fd_op(op: FdOp, proxy_fd: int, arg1: int, arg2: int, data: bytes) -> tuple: |
no test coverage detected