Add TODO.md, ARCHITECTURE.md. Update README.md. and add kernel proxy

This commit is contained in:
xwings
2026-04-11 22:48:18 +08:00
parent fe462b8313
commit 30eba376fc
12 changed files with 2214 additions and 99 deletions
+4
View File
@@ -70,3 +70,7 @@ poetry.toml
# LSP config files
pyrightconfig.json
# AI
CLAUDE.md
.claude
+427
View File
@@ -0,0 +1,427 @@
# Qiling Framework Architecture
Qiling is a cross-platform, multi-architecture binary emulation framework built on top of the [Unicorn](https://www.unicorn-engine.org/) CPU emulation engine. It adds OS-level abstractions (syscalls, file systems, loaders) on top of raw CPU emulation, enabling full binary execution without native hardware.
## High-Level Overview
```
┌──────────────────────────────────────────────────┐
│ User Script / qltool │
├──────────────────────────────────────────────────┤
│ Qiling Core (core.py) │
│ hooks · state snapshots · patches │
├────────────┬─────────────┬───────────────────────┤
│ OS Layer │ Loader │ Memory Manager │
│ (QlOs) │ (QlLoader) │ (QlMemoryManager) │
│ syscalls │ ELF/PE/ │ map · read · write │
│ APIs │ MachO/etc │ MMIO callbacks │
├────────────┴─────────────┴───────────────────────┤
│ Architecture Layer (QlArch) │
│ registers · disassembly · calling conventions │
├──────────────────────────────────────────────────┤
│ Unicorn Engine (CPU) │
│ instruction-level emulation │
└──────────────────────────────────────────────────┘
```
## Directory Structure
```
qiling/
├── qiling/ # Core framework package
│ ├── core.py # Qiling class — main entry point and orchestrator
│ ├── core_hooks.py # Hook system (code, memory, interrupt, address hooks)
│ ├── core_hooks_types.py # Hook type definitions and dispatch
│ ├── core_struct.py # Endian-aware struct packing utilities
│ ├── const.py # Enumerations: QL_ARCH, QL_OS, QL_ENDIAN, etc.
│ ├── exception.py # Custom exception hierarchy
│ ├── utils.py # Component selection (select_arch, select_os, etc.)
│ ├── host.py # Host platform interface
│ ├── log.py # Logging configuration
│ │
│ ├── arch/ # Architecture implementations
│ │ ├── arch.py # QlArch — abstract base class
│ │ ├── x86.py # x86 / x86-64 / 8086
│ │ ├── arm.py # ARMv7 (32-bit)
│ │ ├── arm64.py # ARMv8 (64-bit)
│ │ ├── mips.py # MIPS32
│ │ ├── riscv.py # RISC-V 32-bit
│ │ ├── riscv64.py # RISC-V 64-bit
│ │ ├── ppc.py # PowerPC 32-bit
│ │ ├── cortex_m.py # ARM Cortex-M (MCU)
│ │ ├── register.py # Register management
│ │ └── models.py # CPU model definitions
│ │
│ ├── os/ # Operating system implementations
│ │ ├── os.py # QlOs — abstract base class
│ │ ├── memory.py # QlMemoryManager
│ │ ├── fcall.py # Function call interface (read params, set return)
│ │ ├── mapper.py # Syscall/API mapping
│ │ ├── path.py # Virtual filesystem path resolution
│ │ ├── filestruct.py # File descriptor abstraction
│ │ ├── thread.py # Threading primitives
│ │ ├── posix/ # POSIX shared layer (syscall handlers)
│ │ ├── linux/ # Linux-specific OS
│ │ ├── freebsd/ # FreeBSD-specific OS
│ │ ├── macos/ # macOS-specific OS
│ │ ├── qnx/ # QNX RTOS
│ │ ├── windows/ # Windows (Win32/Win64 API emulation)
│ │ ├── uefi/ # UEFI firmware services
│ │ ├── dos/ # DOS (8086 interrupts)
│ │ ├── mcu/ # Bare-metal microcontroller
│ │ └── blob/ # Raw binary blob execution
│ │
│ ├── loader/ # Binary format loaders
│ │ ├── loader.py # QlLoader — abstract base class
│ │ ├── elf.py # ELF (Linux, FreeBSD, QNX)
│ │ ├── pe.py # PE (Windows)
│ │ ├── pe_uefi.py # PE for UEFI
│ │ ├── macho.py # Mach-O (macOS)
│ │ ├── dos.py # DOS COM/EXE
│ │ ├── mcu.py # MCU firmware images
│ │ └── blob.py # Raw binary blobs
│ │
│ ├── cc/ # Calling conventions
│ │ ├── intel.py # cdecl, stdcall, ms64
│ │ ├── arm.py # aarch32, aarch64
│ │ ├── mips.py # MIPS o32
│ │ ├── riscv.py # RISC-V ABI
│ │ └── ppc.py # PowerPC ABI
│ │
│ ├── hw/ # Hardware peripheral emulation (MCU)
│ │ ├── peripheral.py # Base peripheral class
│ │ ├── hw.py # Hardware manager
│ │ ├── gpio/ # GPIO pins and interrupts
│ │ ├── timer/ # Timers, PWM, counters
│ │ ├── char/ # UART serial
│ │ ├── spi/ # SPI bus
│ │ ├── i2c/ # I2C bus
│ │ ├── net/ # Network interfaces
│ │ ├── analog/ # ADC/DAC
│ │ ├── intc/ # Interrupt controllers
│ │ ├── flash/ # Flash memory
│ │ ├── dma/ # DMA controllers
│ │ └── ... # Power, SD, misc peripherals
│ │
│ ├── debugger/ # Debugger subsystem
│ │ ├── gdb/ # GDB remote protocol server
│ │ └── qdb/ # Qiling native debugger (with reverse debugging)
│ │
│ ├── extensions/ # Optional extensions
│ │ ├── multitask.py # gevent-based multithreading
│ │ ├── trace.py # Instruction tracing
│ │ ├── coverage/ # Code coverage collection
│ │ ├── sanitizers/ # Memory sanitizers
│ │ ├── afl/ # AFL fuzzer integration
│ │ ├── r2/ # Radare2 integration
│ │ └── idaplugin/ # IDA Pro plugin
│ │
│ └── profiles/ # Default OS configuration files (.ql)
│ ├── linux.ql # Stack/heap addresses, kernel params
│ ├── windows.ql
│ ├── macos.ql
│ └── ...
│
├── qltool # CLI tool for running binaries
├── qltui.py # TUI interface
├── examples/ # Usage examples and sample scripts
├── tests/ # Test suite
└── docs/ # Documentation
```
## Core Components
### `Qiling` (core.py)
The central class. Every emulation session creates one `Qiling` instance that owns and wires together all other components:
```python
ql = Qiling(
argv=["/path/to/binary", "arg1"], # binary + arguments
rootfs="/path/to/rootfs", # virtual filesystem root
ostype=QL_OS.LINUX, # target OS (auto-detected if omitted)
archtype=QL_ARCH.X8664, # target arch (auto-detected if omitted)
)
```
`Qiling` inherits from `QlCoreHooks` (hook management) and `QlCoreStructs` (endian-aware packing). Key properties:
| Property | Type | Description |
|----------|------|-------------|
| `ql.arch` | `QlArch` | CPU architecture — registers, disassembly |
| `ql.os` | `QlOs` | Operating system — syscalls, I/O, APIs |
| `ql.loader` | `QlLoader` | Binary loader — parses and maps the executable |
| `ql.mem` | `QlMemoryManager` | Memory — map, read, write, MMIO |
| `ql.uc` | `unicorn.Uc` | Underlying Unicorn engine instance |
### Architecture Layer (`qiling/arch/`)
`QlArch` is the abstract base. Each architecture subclass configures:
- **Unicorn engine** mode and architecture constants
- **Register** access via `ql.arch.regs` (read/write by name)
- **Disassembler** (Capstone) and **assembler** (Keystone)
- **Stack operations** — push, pop, pointer-width-aware
- **Endianness** and **bit width** (16/32/64)
Supported: x86, x86-64, 8086, ARM, ARM64, MIPS, RISC-V (32/64), PowerPC, Cortex-M.
### OS Layer (`qiling/os/`)
`QlOs` is the abstract base. Each OS subclass provides:
- **Syscall/interrupt dispatch** — routes CPU interrupts to handler functions
- **I/O streams** — `stdin`, `stdout`, `stderr` (interceptable)
- **Virtual filesystem** — path mapping through `rootfs`
- **Function call interface** (`ql.os.fcall`) — read params, set return values
- **API interception** — `set_api()` for hooking library functions
**POSIX subsystem** (`os/posix/`): Shared syscall implementation for Linux, FreeBSD, macOS, and QNX. Individual syscall handlers live under `os/posix/syscall/`.
**Windows** (`os/windows/`): Emulates Win32/Win64 API by hooking DLL imports. Includes registry, thread, handle, and fiber support.
**UEFI** (`os/uefi/`): Emulates UEFI Boot Services, Runtime Services, and SMM. Uses a GUID database and protocol framework.
### Loader Layer (`qiling/loader/`)
`QlLoader` is the abstract base. Loaders parse a binary format, map segments into memory, resolve symbols, load dependencies, and set initial CPU state (PC, SP).
| Loader | Format | Used By |
|--------|--------|---------|
| `QlLoaderELF` | ELF | Linux, FreeBSD, QNX |
| `QlLoaderPE` | PE/COFF | Windows, UEFI |
| `QlLoaderMacho` | Mach-O | macOS |
| `QlLoaderDOS` | COM/EXE | DOS |
| `QlLoaderMCU` | Firmware | Cortex-M MCU |
| `QlLoaderBlob` | Raw bytes | Shellcode / blob |
### Memory Manager (`qiling/os/memory.py`)
Wraps Unicorn's memory model with higher-level operations:
- `map(addr, size, perms)` / `unmap(addr, size)` — region management
- `read(addr, size)` / `write(addr, data)` — data access
- `read_ptr(addr)` / `write_ptr(addr, val)` — pointer-width-aware access
- `read_cstring(addr)` — null-terminated string read
- MMIO callback support for memory-mapped peripherals
### Calling Conventions (`qiling/cc/`)
Each architecture has calling convention classes that abstract argument passing and return values. The `QlOs.fcall` interface uses these to provide a uniform way to read function parameters regardless of platform.
## Execution Flow
### 1. Initialization (`Qiling.__init__`)
```
Qiling(argv, rootfs)
│
├─ Detect arch/OS from binary headers (ql_guess_emu_env)
│ ELF magic → parse e_machine, OSABI
│ PE magic → parse Machine, Subsystem
│ MachO magic → parse CPU type
│
├─ Create QlArch (select_arch) → initializes Unicorn engine
├─ Create QlLoader (select_loader)
├─ Create QlMemoryManager
├─ Create QlOs (select_os)
│
└─ loader.run()
├─ Parse binary format (headers, segments, sections)
├─ Map segments into memory
├─ Load shared libraries / DLLs
├─ Setup stack, heap, TLS, auxiliary vectors
└─ Set initial PC (entry point) and SP
```
### 2. Execution (`ql.run()`)
```
ql.run(begin, end, timeout, count)
│
├─ Apply binary patches (ql.patch)
├─ Write exit trap (guard address)
│
└─ os.run()
└─ uc.emu_start(entry_point, exit_point)
│
├─ Unicorn executes instructions
│
├─ Hooks fire on:
│ ├─ Every instruction (hook_code)
│ ├─ Basic blocks (hook_block)
│ ├─ Memory access (hook_mem_read/write)
│ ├─ Interrupts (hook_intno) → syscall dispatch
│ ├─ Specific addresses (hook_address)
│ └─ Specific instructions (hook_insn)
│
└─ Stops when PC reaches exit point, timeout,
ql.emu_stop(), or unhandled exception
```
### 3. Syscall Handling
When the emulated binary issues a syscall (via `int 0x80`, `syscall`, `svc`, etc.):
```
CPU interrupt/instruction
→ Unicorn interrupt hook
→ QlOs syscall dispatcher
→ Look up handler by syscall number
→ Handler reads args via calling convention
→ Emulates syscall behavior
→ Sets return value in registers
```
## Component Selection
Components are selected dynamically at runtime based on `QL_ARCH` and `QL_OS` enums. The `qiling/utils.py` module provides:
- `select_arch(archtype)` → architecture class
- `select_os(ostype)` → OS class
- `select_loader(ostype)` → loader class
- `select_debugger(options)` → debugger class
This makes it possible to support diverse platform combinations from a unified codebase.
## Hook System
The hook system (`core_hooks.py`) wraps Unicorn's callback mechanism:
| Hook Type | Trigger |
|-----------|---------|
| `hook_code` | Every instruction (optionally within address range) |
| `hook_block` | Every basic block entry |
| `hook_address` | Specific address reached |
| `hook_intno` | CPU interrupt/exception |
| `hook_insn` | Specific instruction type (e.g., `syscall`) |
| `hook_mem_read` | Memory read |
| `hook_mem_write` | Memory write |
| `hook_mem_invalid` | Invalid memory access |
Hooks can be scoped to address ranges and return `QL_HOOK_BLOCK` to suppress further hooks in the chain.
## Key Extension Points
- **Custom syscall handlers** — replace or extend any syscall
- **API hooking** — `ql.os.set_api(name, callback)` to intercept library calls
- **Binary patching** — `ql.patch(offset, data)` for runtime patching
- **State snapshots** — `ql.save()` / `ql.restore()` for checkpointing
- **Debugger attachment** — GDB remote protocol or native QDB debugger
- **Coverage/tracing** — `extensions/coverage/` and `extensions/trace.py`
- **Fuzzing** — AFL integration via `extensions/afl/`
- **Hardware peripherals** — register custom MCU peripherals in `hw/`
## Dependencies
| Package | Role |
|---------|------|
| `unicorn` (2.1.3) | CPU emulation engine |
| `capstone` | Disassembly |
| `keystone-engine` | Assembly |
| `pyelftools` | ELF parsing |
| `pefile` | PE parsing |
| `python-registry` | Windows registry emulation |
| `gevent` | Cooperative multithreading |
| `pyyaml` | Configuration parsing |
Optional: `unicornafl` / `fuzzercorn` (fuzzing), `r2libr` (Radare2 integration).
## Supported Platforms
**Architectures:** x86, x86-64, 8086, ARM, ARM64, MIPS, RISC-V (32/64), PowerPC, Cortex-M
**Operating Systems:** Linux, FreeBSD, macOS, Windows, UEFI, DOS, QNX, MCU (bare-metal), Blob
## Improvement: Hybrid Kernel Architecture
> Detailed implementation plan and task tracking: [TODO.md](TODO.md)
### The Problem
Qiling reimplements Linux kernel behavior syscall-by-syscall in Python. This works
for simple operations (file I/O, memory management, stat) but fundamentally cannot
scale to the full kernel surface:
- **Networking**: No epoll. Sockets are proxied to host sockets with no isolation.
No real TCP state machine, no multicast, no raw/netlink sockets.
- **Multithreading**: Gevent greenlets are cooperative and single-threaded. No
preemption, no real concurrency. Futex is a gevent Event. Programs using pthreads,
mutexes, or condition variables don't behave correctly.
- **Signals**: `signal()`, `sigaction()`, `kill()` are mostly stubbed. No delivery,
no `EINTR`, no `SA_RESTART`.
- **Long tail**: capabilities, cgroups, namespaces, io_uring, seccomp, eBPF — the
kernel API surface is vast and growing.
### The Solution
A **hybrid architecture** that keeps Unicorn for CPU emulation and Qiling for
instrumentation, but offloads complex kernel subsystems to a real Linux kernel via
a **kernel proxy** helper process. Simple syscalls stay emulated in Python.
```
Syscall interrupt
→ load_syscall() [UNCHANGED — existing dispatch in posix.py]
→ check posix_syscall_hooks[CALL]
→ proxy hook registered? → forward to kernel proxy
→ no proxy hook? → existing Python handler [UNCHANGED]
```
The user explicitly chooses which missing syscalls to forward. Nothing is automatic —
by default Qiling behaves exactly as today. The integration uses the **existing
`set_syscall()` CALL hook mechanism** (`posix.py:128-143`), so `load_syscall()` and
all existing dispatch code remain completely unchanged.
```python
proxy = KernelProxy(ql)
proxy.forward_syscall("epoll_create", returns_fd=True)
proxy.forward_syscall("epoll_ctl")
proxy.forward_syscall("epoll_wait")
ql.run()
```
Under the hood, `forward_syscall()` registers a CALL hook that serializes the
arguments and sends them to a helper process (the kernel proxy) which executes
the real syscall and returns the result. For syscalls that return FDs, the result
is wrapped in a `ql_proxy_fd` object and stored in Qiling's FD table. Since the
FD table is already polymorphic (`ql_socket`, `ql_file`, `ql_pipe`), existing
handlers like `ql_syscall_read` and `ql_syscall_close` dispatch through the proxy
FD's `.read()`/`.close()` methods automatically — no changes needed.
### Phases
| Phase | Scope | Risk | Goal |
|-------|-------|------|------|
| 0 | Proof of concept | Low | User manually forwards specific syscalls — zero existing code changed |
| 1 | Networking foundation | Low-Med | Specific hooks for socket syscalls, `ql_proxy_fd`, TCP works |
| 2 | Complete networking | Medium | epoll, poll/select, network namespaces |
| 3 | Real threading | **High** | One Unicorn per thread, shared memory, real futex |
| 4 | Signals | Medium | Real signal delivery, EINTR, handler execution |
| 5 | Integration | Low | API polish, fallback, platform support, benchmarks |
Phase 0 gives users explicit control — they identify which missing syscalls to
forward and the proxy handles them. Phases 1-2 add pointer-aware forwarding for
networking with pre-built forwarders so users don't have to wire up each syscall.
Phase 3 (threading) is the highest-risk change and is deferred until networking is
stable. Each phase preserves backward compatibility — hybrid mode is opt-in, default
behavior is unchanged.
### Alternatives Considered
- **Run a real kernel in Unicorn**: Unicorn doesn't emulate hardware (interrupt
controllers, MMU page tables, timers). Would require rebuilding QEMU system mode.
- **ptrace-based execution**: Run natively, intercept syscalls. Fast, but no
cross-architecture support and limited instruction-level hooks.
- **User-Mode Linux (UML)**: Run the kernel as a userspace process. x86-only,
somewhat unmaintained, complex syscall bridge.
- **Auto-forward all unimplemented syscalls**: Forward every missing syscall
automatically. Convenient but unpredictable — hard to debug, may forward syscalls
that shouldn't be (security, state leaks). Explicit user control is safer.
The hybrid approach was chosen because it preserves Qiling's core value
(instrumentation + cross-arch emulation) while getting real kernel behavior where
it matters most — without modifying the existing dispatch path.
## Testing
Tests live in `tests/` and are organized by platform: `test_elf.py`, `test_pe.py`, `test_macho.py`, `test_dos.py`, `test_mcu.py`, etc. They use binaries from `examples/rootfs/` as test fixtures.
+101 -97
View File
@@ -1,4 +1,4 @@
[![Documentation Status](https://github.com/qilingframework/qiling/wiki)](https://github.com/qilingframework/qiling/wiki)
[![Documentation Status](https://readthedocs.org/projects/qilingframework/badge/?version=latest)](https://docs.qiling.io)
[![Downloads](https://pepy.tech/badge/qiling)](https://pepy.tech/project/qiling)
[![Chat on Telegram](https://img.shields.io/badge/Chat%20on-Telegram-brightgreen.svg)](https://t.me/qilingframework)
@@ -8,43 +8,24 @@
<img width="150" height="150" src="https://raw.githubusercontent.com/qilingframework/qiling/master/docs/qiling2_logo_small.png">
</p>
# Qiling Framework
[Qiling's use case, blog and related work](https://github.com/qilingframework/qiling/issues/134)
Qiling is an advanced binary emulation framework that allows you to emulate and sandbox code in an isolated environment across multiple platforms and architectures. Built on top of Unicorn Engine, Qiling provides a higher-level framework that understands operating system contexts, executable formats, and dynamic linking.
Qiling is an advanced binary emulation framework, with the following features:
## Table of Contents
- [Features](#features)
- [Appearance](#Appearance)
- [Use Cases](#use-cases)
- [Quick Start](#quick-start)
- [Installation](#installation)
- [Basic Usage](#basic-usage)
- [Qiling vs. Other Emulators](#qiling-vs-other-emulators)
- [Qiling vs. Unicorn Engine](#qiling-vs-unicorn-engine)
- [Qiling vs. QEMU User Mode](#qiling-vs-qemu-user-mode)
- [Examples](#examples)
- [Qltool](#qltool)
- [Contributing](#contributing)
- [License](#license)
- [Contact](#contact)
- [Core Developers & Contributors](#core-developers--contributors)
## Features
- **Multi-platform Emulation**: Windows, macOS, Linux, Android, BSD, UEFI, DOS, MBR.
- **Multi-architecture Emulation**: 8086, X86, X86_64, ARM, ARM64, MIPS, RISC-V, PowerPC.
- **Multiple File Format Support**: PE, Mach-O, ELF, COM, MBR.
- **Kernel Module Emulation**: Supports Windows Driver (.sys), Linux Kernel Module (.ko) & macOS Kernel (.kext) via [Demigod](https://groundx.io/demigod/).
- **Isolated Sandboxing**: Emulates & sandboxes code in an isolated environment with a fully configurable sandbox.
- **In-depth API**: Provides in-depth memory, register, OS level, and filesystem level API.
- **Fine-grain Instrumentation**: Allows hooks at various levels (instruction/basic-block/memory-access/exception/syscall/IO/etc.).
- **Virtual Machine Level API**: Supports saving and restoring the current execution state.
- **Debugging Capabilities**: Supports cross-architecture and platform debugging, including a built-in debugger with reverse debugging capability.
- **Dynamic Hot Patching**: Allows dynamic hot patching of on-the-fly running code, including loaded libraries.
- **Python Framework**: A true framework in Python, making it easy to build customized security analysis tools.
## Appearance
- Emulate multi-platforms: Windows, macOS, Linux, Android, BSD, UEFI, DOS, MBR.
- Emulate multi-architectures: 8086, X86, X86_64, ARM, ARM64, MIPS, RISC-V, PowerPC.
- Support multiple file formats: PE, Mach-O, ELF, COM, MBR.
- Support Windows Driver (.sys), Linux Kernel Module (.ko) & macOS Kernel (.kext) via [Demigod](https://groundx.io/demigod/).
- Emulates & sandbox code in an isolated environment.
- Provides a fully configurable sandbox.
- Provides in-depth memory, register, OS level and filesystem level API.
- Fine-grain instrumentation: allows hooks at various levels
(instruction/basic-block/memory-access/exception/syscall/IO/etc.)
- Provides virtual machine level API such as saving and restoring the current execution state.
- Supports cross architecture and platform debugging capabilities.
- Built-in debugger with reverse debugging capability.
- Allows dynamic hot patch on-the-fly running code, including the loaded library.
- True framework in Python, making it easy to build customized security analysis tools on top.
Qiling also made its way to various international conferences.
@@ -68,37 +49,79 @@ Qiling also made its way to various international conferences.
- [Nullcon](https://nullcon.net/website/goa-2020/speakers/kaijern-lau.php)
2019:
- [DEFCON, USA](https://www.defcon.org/html/defcon-27/dc-27-demolabs.html#QiLing)
- [Hitcon](https://hitcon.org/2019/CMT/agenda)
- [Zeronights](https://zeronights.ru/report-en/qiling-io-advanced-binary-emulation-framework/)
## Use Cases
Qiling has been presented at various international conferences, showcasing its versatility in:
Qiling is backed by [Unicorn Engine](http://www.unicorn-engine.org).
- Binary analysis and reverse engineering.
- Malware analysis and sandboxing.
- Firmware analysis and emulation.
- Security research and vulnerability discovery.
- CTF challenges and exploit development.
Visit our [website](https://www.qiling.io) for more information.
For more details on Qiling's use cases, blog posts, and related work, please refer to [Qiling's use case, blog and related work](https://github.com/qilingframework/qiling/issues/134).
---
#### License
## Quick Start
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
### Installation
---
Qiling requires Python 3.8 or newer. You can install it using pip:
#### Qiling vs. other Emulators
```bash
pip install qiling
```
There are many open-source emulators, but two projects closest to Qiling
are [Unicorn](http://www.unicorn-engine.org) & [QEMU user mode](https://qemu.org).
This section explains the main differences of Qiling against them.
For more detailed installation instructions and dependencies, please refer to the [official documentation](https://github.com/qilingframework/qiling/wiki/Installation).
##### Qiling vs. Unicorn engine
### Basic Usage
Built on top of Unicorn, but Qiling & Unicorn are two different animals.
The example below shows how to use Qiling framework in the most straightforward way to emulate a Windows executable.
- Unicorn is just a CPU emulator, so it focuses on emulating CPU instructions,
that can understand emulator memory.
Beyond that, Unicorn is not aware of higher level concepts, such as dynamic
libraries, system calls, I/O handling or executable formats like PE, Mach-O
or ELF. As a result, Unicorn can only emulate raw machine instructions,
without Operating System (OS) context.
- Qiling is designed as a higher level framework, that leverages Unicorn to
emulate CPU instructions, but can understand OS: it has executable format
loaders (for PE, Mach-O & ELF currently), dynamic linkers (so we can
load & relocate shared libraries), syscall & IO handlers. For this reason,
Qiling can run executable binary without requiring its native OS.
##### Qiling vs. QEMU user mode
QEMU user mode does a similar thing to our emulator, that is, to emulate whole
executable binaries in a cross-architecture way.
However, Qiling offers some important differences against QEMU user mode:
- Qiling is a true analysis framework,
that allows you to build your own dynamic analysis tools on top (in Python).
Meanwhile, QEMU is just a tool, not a framework.
- Qiling can perform dynamic instrumentation, and can even hot patch code at
runtime. QEMU does neither.
- Not only working cross-architecture, Qiling is also cross-platform.
For example, you can run Linux ELF file on top of Windows.
In contrast, QEMU user mode only runs binary of the same OS, such as Linux
ELF on Linux, due to the way it forwards syscall from emulated code to
native OS.
- Qiling supports more platforms, including Windows, macOS, Linux & BSD. QEMU
user mode can only handle Linux & BSD.
---
#### Installation
Please see [setup guide](https://github.com/qilingframework/qiling/wiki/Installation) file for how to install Qiling Framework.
---
#### Examples
The example below shows how to use Qiling framework in the most
straightforward way to emulate a Windows executable.
```python
from qiling import Qiling
@@ -112,30 +135,8 @@ if __name__ == "__main__":
ql.run()
```
## Qiling vs. Other Emulators
There are many open-source emulators, but two projects closest to Qiling are [Unicorn](http://www.unicorn-engine.org) & [QEMU user mode](https://qemu.org). This section explains the main differences of Qiling against them.
### Qiling vs. Unicorn Engine
Built on top of Unicorn, but Qiling & Unicorn are two different animals.
- **Unicorn** is just a CPU emulator, so it focuses on emulating CPU instructions, that can understand emulator memory. Beyond that, Unicorn is not aware of higher level concepts, such as dynamic libraries, system calls, I/O handling or executable formats like PE, Mach-O or ELF. As a result, Unicorn can only emulate raw machine instructions, without Operating System (OS) context.
- **Qiling** is designed as a higher level framework, that leverages Unicorn to emulate CPU instructions, but can understand OS: it has executable format loaders (for PE, Mach-O & ELF currently), dynamic linkers (so we can load & relocate shared libraries), syscall & IO handlers. For this reason, Qiling can run executable binary without requiring its native OS.
### Qiling vs. QEMU User Mode
QEMU user mode does a similar thing to our emulator, that is, to emulate whole executable binaries in a cross-architecture way.
However, Qiling offers some important differences against QEMU user mode:
- **Qiling is a true analysis framework**, that allows you to build your own dynamic analysis tools on top (in Python). Meanwhile, QEMU is just a tool, not a framework.
- **Qiling can perform dynamic instrumentation**, and can even hot patch code at runtime. QEMU does neither.
- Not only working cross-architecture, **Qiling is also cross-platform**. For example, you can run Linux ELF file on top of Windows. In contrast, QEMU user mode only runs binary of the same OS, such as Linux ELF on Linux, due to the way it forwards syscall from emulated code to native OS.
- **Qiling supports more platforms**, including Windows, macOS, Linux & BSD. QEMU user mode can only handle Linux & BSD.
## Examples
- The following example shows how a Windows crackme may be patched dynamically to make it always display the “Congratulation” dialog.
- The following example shows how a Windows crackme may be patched dynamically
to make it always display the “Congratulation” dialog.
```python
from qiling import Qiling
@@ -176,13 +177,15 @@ The below YouTube video shows how the above example works.
#### Emulating ARM router firmware on Ubuntu x64 host
Qiling Framework hot-patches and emulates an ARM router's `/usr/bin/httpd` on an x86_64 Ubuntu host.
Qiling Framework hot-patches and emulates an ARM router's `/usr/bin/httpd` on
an x86_64 Ubuntu host.
[![Qiling Tutorial: Emulating and Fuzz ARM router firmware](https://github.com/qilingframework/theme.qiling.io/blob/master/source/img/fuzzer.jpg?raw=true)](https://www.youtube.com/watch?v=e3_T3KLhNUs)
[![Qiling Tutorial: Emulating and Fuzz ARM router firmware](https://github.com/qilingframework/theme.qiling.io/blob/master/source/img/fuzzer.jpg?raw=true)](https://www.youtube.com/watch?v=e3_T3KLh2NU)
#### Qiling's IDA Pro Plugin: Instrument and Decrypt Mirai's Secret
This video demonstrates how Qiling's IDA Pro plugin can make IDA Pro run with Qiling instrumentation engine.
This video demonstrates how Qiling's IDA Pro plugin can make IDA Pro run with
Qiling instrumentation engine.
[![Qiling's IDA Pro Plugin: Instrument and Decrypt Mirai's Secret](http://img.youtube.com/vi/ZWMWTq2WTXk/0.jpg)](http://www.youtube.com/watch?v=ZWMWTq2WTXk)
@@ -192,62 +195,63 @@ Solving a simple CTF challenge with Qiling Framework and IDA Pro
[![Solving a simple CTF challenge with Qiling Framework and IDA Pro](https://i.ytimg.com/vi/SPjVAt2FkKA/0.jpg)](https://www.youtube.com/watch?v=SPjVAt2FkKA)
#### Emulating MBR
Qiling Framework emulates MBR
[![Qiling DEMO: Emulating MBR](https://github.com/qilingframework/theme.qiling.io/blob/master/source/img/mbr.png?raw=true)](https://github.com/qilingframework/theme.qiling.io/blob/master/source/img/mbr.png?raw=true)
## Qltool
---
#### Qltool
Qiling also provides a friendly tool named `qltool` to quickly emulate shellcode & executable binaries.
With qltool, easy execution can be performed:
With shellcode:
```bash
```
$ ./qltool code --os linux --arch arm --format hex -f examples/shellcodes/linarm32_tcp_reverse_shell.hex
```
With binary file:
```bash
```
$ ./qltool run -f examples/rootfs/x8664_linux/bin/x8664_hello --rootfs examples/rootfs/x8664_linux/
```
With binary and GDB debugger enabled:
```bash
```
$ ./qltool run -f examples/rootfs/x8664_linux/bin/x8664_hello --gdb 127.0.0.1:9999 --rootfs examples/rootfs/x8664_linux
```
With code coverage collection (UEFI only for now):
```bash
```
$ ./qltool run -f examples/rootfs/x8664_efi/bin/TcgPlatformSetupPolicy --rootfs examples/rootfs/x8664_efi --coverage-format drcov --coverage-file TcgPlatformSetupPolicy.cov
```
With JSON output (Windows, mainly):
```bash
```
$ ./qltool run -f examples/rootfs/x86_windows/bin/x86_hello.exe --rootfs examples/rootfs/x86_windows/ --console False --json
```
---
## Contributing
We welcome contributions from the community! If you're interested in contributing to Qiling Framework, please check out our [GitHub repository](https://github.com/qilingframework/qiling) and look for open issues or submit a pull request.
#### Contact
## License
Get the latest info from our website https://www.qiling.io
This program is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version.
Contact us at email info@qiling.io,
via Twitter [@qiling_io](https://twitter.com/qiling_io).
## Contact
---
Get the latest info from our website [https://www.qiling.io](https://www.qiling.io)
#### Core developers, Key Contributors and etc.
Contact us at email [[email protected]](mailto:[email protected]), or via Twitter [@qiling_io](https://twitter.com/qiling_io).
## Core Developers & Contributors
Please refer to [CREDITS.md](https://github.com/qilingframework/qiling/blob/dev/CREDITS.md).
Please refer to [CREDITS.md](https://github.com/qilingframework/qiling/blob/dev/CREDITS.md).
-1
View File
@@ -1 +0,0 @@
Features request and TODO please refer to issue 333 https://github.com/qilingframework/qiling/issues/333
+693
View File
@@ -0,0 +1,693 @@
# Qiling Framework TODO
Features request and TODO please refer to issue 333 https://github.com/qilingframework/qiling/issues/333
---
## Hybrid Kernel Architecture
### The Problem
Qiling reimplements Linux kernel behavior syscall-by-syscall in Python. Simple
syscalls (file I/O, memory, stat) work well. Complex subsystems do not:
- **Networking**: No epoll. Sockets are host sockets with no isolation. No TCP
state machine, no multicast, no raw/netlink sockets.
- **Threading**: Gevent greenlets — cooperative, single-threaded. No preemption.
Futex is a gevent Event. pthreads don't behave correctly.
- **Signals**: `signal()`, `sigaction()`, `kill()` are stubs. No delivery, no EINTR.
Reimplementing the full kernel is not realistic. Instead, offload complex subsystems
to a real Linux kernel while keeping Qiling's instrumentation intact.
### Key Insight: Two-Layer Forwarding
There are two integration points, each serving a different purpose:
**Layer 1 — Generic fallback** (catches the long tail):
The user explicitly chooses which missing syscalls to forward. Nothing is automatic.
By default Qiling behaves exactly as it does today — if a syscall is not implemented,
it fails. The user then tells the proxy which specific syscalls to forward:
```python
ql = Qiling(argv=["/bin/myserver"], rootfs="rootfs/x8664_linux")
proxy = KernelProxy(ql) # start proxy process
proxy.forward_syscall("epoll_create") # forward this one to real kernel
proxy.forward_syscall("epoll_ctl") # and this one
proxy.forward_syscall("epoll_wait") # and this one
ql.run()
```
Under the hood, `forward_syscall("epoll_create")` registers a CALL hook via the
existing `set_syscall()` mechanism (`posix.py:128-143`). The hook reads args from
the emulated registers, sends them to the proxy process, and returns the real result.
**Zero changes to `load_syscall()` or any existing dispatch code.**
For syscalls that return FDs (socket, epoll_create, eventfd, etc.), the proxy
wraps the returned FD in a `ql_proxy_fd` object and stores it in the FD table.
The FD table (`QlFileDes`) already stores polymorphic objects — `ql_socket`,
`ql_file`, `ql_pipe` — so `ql_proxy_fd` slots in naturally. When existing handlers
like `ql_syscall_read` or `ql_syscall_close` hit a proxy FD, they dispatch through
`ql_proxy_fd.read()` / `.close()` which forwards to the proxy.
```
Syscall interrupt
→ load_syscall() [UNCHANGED]
→ has CALL hook? (user hook or proxy-registered hook)
→ yes: use it
→ has Python handler? (existing code)
→ yes: use it [unchanged — file I/O, memory, stat, etc.]
→ neither?
→ log warning [existing behavior, unchanged — no auto-forwarding]
```
### What Changes vs What Doesn't
| Component | Changes? | Notes |
|-----------|----------|-------|
| `load_syscall()` dispatch | **NO** | Entirely unchanged |
| Existing syscall handlers | NO | Python handlers stay as-is |
| `QlFileDes` FD table | NO | Already polymorphic, new FD type slots in |
| `set_syscall()` user hooks | NO | User CALL/ENTER/EXIT hooks still work |
| `ql.run()` / `ql.emu_start()` | NO | Unicorn execution loop untouched |
| Hook system (`core_hooks.py`) | NO | Standard Unicorn hook mechanism |
| New: kernel proxy process | YES | New module, new files only |
| New: `ql_proxy_fd` FD type | YES | New class, same interface as `ql_socket` |
| New: `KernelProxy` class | YES | User-facing API, registers CALL hooks |
---
## Phase 0: Proof of Concept
**Goal**: User can explicitly forward specific unimplemented syscalls to a real Linux
kernel. No automatic behavior. No changes to existing Qiling dispatch code.
### Usage
```python
from qiling import Qiling
from qiling.os.posix.kernel_proxy import KernelProxy
ql = Qiling(argv=["/bin/myserver"], rootfs="rootfs/x8664_linux")
# Start a kernel proxy — a helper process that executes real syscalls
proxy = KernelProxy(ql)
# Binary needs epoll but Qiling doesn't implement it.
# User identifies the 3 missing syscalls and forwards them:
proxy.forward_syscall("epoll_create")
proxy.forward_syscall("epoll_ctl")
proxy.forward_syscall("epoll_wait")
ql.run()
# epoll_create/ctl/wait are handled by real kernel.
# Everything else (read, write, open, mmap, ...) uses existing Qiling handlers.
```
If the user does NOT set up a proxy, Qiling behaves exactly as it does today.
No surprises, no magic.
### 0.1 Kernel proxy process
A standalone Python process that executes real Linux syscalls on behalf of Qiling.
Communicates via Unix socketpair.
- [ ] New directory: `qiling/os/posix/kernel_proxy/`
- [ ] `proxy.py` — proxy subprocess entry point
- Main loop: read request from socket → `libc.syscall(nr, *args)` → write response
- Uses `ctypes.CDLL("libc.so.6").syscall()` for raw syscall execution
- Manages its own FD table (proxy-side FDs)
- [ ] IPC protocol (binary, over socketpair):
```
Request: { type: SYSCALL, syscall_nr: u32, args: [u64; 6] }
Response: { return_value: i64, errno: i32 }
Request: { type: FD_OP, op: READ|WRITE|CLOSE, proxy_fd: i32, length: u32, data?: bytes }
Response: { return_value: i64, errno: i32, data?: bytes }
```
Two message types: raw syscall forwarding, and FD operations (for read/write/close
on proxy-owned FDs).
- [ ] Lifecycle: started by `KernelProxy.__init__()`, killed on `ql.run()` exit
### 0.2 `KernelProxy` class — user-facing API
The main integration class. Lives in `qiling/os/posix/kernel_proxy/__init__.py`.
```python
class KernelProxy:
def __init__(self, ql: Qiling):
"""Start the proxy subprocess."""
self.ql = ql
self._proxy_process = ... # start subprocess
self._ipc = ... # socketpair connection
self._forwarded = {} # syscall_name → syscall_nr
def forward_syscall(self, name: str, returns_fd: bool = False):
"""Register a CALL hook that forwards this syscall to the proxy.
Args:
name: syscall name (e.g. "epoll_create", "eventfd")
returns_fd: if True, wrap the return value in ql_proxy_fd
and store in the FD table
"""
# Look up syscall number from the architecture's syscall table
# Register a CALL hook via ql.os.set_syscall()
ql.os.set_syscall(name, self._make_forwarder(name, returns_fd))
def _make_forwarder(self, name, returns_fd):
"""Create a CALL hook function for this syscall."""
syscall_nr = self._resolve_syscall_nr(name)
def forwarder(ql, *args):
# Send all args as raw integers to proxy
retval = self._ipc.forward(syscall_nr, args)
if returns_fd and retval >= 0:
# Wrap proxy FD and store in Qiling's FD table
proxy_fd = ql_proxy_fd(self._ipc, retval)
guest_fd = self._alloc_fd(ql, proxy_fd)
return guest_fd
return retval
return forwarder
```
Key points:
- `forward_syscall()` uses the existing `set_syscall()` mechanism — a standard
CALL hook. No special dispatch path, no changes to `load_syscall()`.
- User explicitly passes `returns_fd=True` for FD-returning syscalls. No heuristics.
- User ENTER/EXIT hooks still fire around the forwarded syscall (existing behavior
of the hook chain in `load_syscall()` lines 206-224).
- [ ] Implement `KernelProxy` class
- [ ] Implement `_resolve_syscall_nr()` — look up syscall number from arch tables
- [ ] Implement `_make_forwarder()` — create CALL hook closure
- [ ] Implement `_alloc_fd()` — find empty slot in `ql.os.fd[]`, store `ql_proxy_fd`
### 0.3 `ql_proxy_fd` — proxy-side FD wrapper
When a forwarded syscall returns an FD (e.g., `epoll_create` returns 5 in the proxy),
we store a `ql_proxy_fd` in Qiling's FD table. This object forwards read/write/close
to the proxy, matching the interface of `ql_socket` (`filestruct.py:14`).
```python
class ql_proxy_fd:
"""FD whose real file/socket lives in the kernel proxy process."""
def __init__(self, ipc, proxy_fd: int):
self._ipc = ipc
self._proxy_fd = proxy_fd
def read(self, length: int) -> bytes:
return self._ipc.fd_read(self._proxy_fd, length)
def write(self, data: bytes) -> int:
return self._ipc.fd_write(self._proxy_fd, data)
def close(self) -> None:
self._ipc.fd_close(self._proxy_fd)
def fileno(self) -> int:
return -1 # not a real host FD
```
Because `ql_syscall_read` / `ql_syscall_write` / `ql_syscall_close` already dispatch
through `ql.os.fd[fd].read()` / `.write()` / `.close()`, **these existing handlers
need no changes**. When the binary calls `read(fd, buf, n)` on a proxy FD, the
existing read handler calls `ql_proxy_fd.read(n)`, gets data back, and writes it
to guest memory as usual.
- [ ] Implement `ql_proxy_fd` class in `qiling/os/posix/kernel_proxy/proxy_fd.py`
- [ ] Verify `ql_syscall_read` works with `ql_proxy_fd` — no changes needed
- [ ] Verify `ql_syscall_write` works with `ql_proxy_fd` — no changes needed
- [ ] Verify `ql_syscall_close` works with `ql_proxy_fd` — no changes needed
### 0.4 Pointer-bearing syscalls (Phase 0 scope: one example)
Some forwarded syscalls take pointers to guest memory. The proxy can't read guest
memory directly. For Phase 0, implement ONE pointer-bearing forwarder as an example
to prove the pattern works. `epoll_ctl` is a good candidate:
```
epoll_ctl(epfd, op, fd, struct epoll_event *event)
```
The forwarder must:
1. Read `struct epoll_event` (8 bytes) from guest memory at the `event` pointer
2. Send the struct data along with the integer args to the proxy
3. Proxy reconstructs the struct in its own memory, calls real `epoll_ctl`
- [ ] Extend IPC protocol for buffer-carrying requests:
```
Request: { type: SYSCALL_WITH_BUFS, syscall_nr, args[6], buffers: [(arg_idx, direction, data)] }
Response: { return_value, errno, buffers: [(arg_idx, data)] }
```
`direction` is IN (guest→proxy), OUT (proxy→guest), or INOUT.
- [ ] Implement `forward_syscall_with_buffers()` API for pointer-bearing syscalls
- [ ] Implement `epoll_ctl` forwarder as the working example
### 0.5 Validation
- [ ] Test: binary that uses `epoll_create` + `epoll_ctl` + `epoll_wait` on a
timerfd or eventfd. User forwards all 4 syscalls. Verify it works end-to-end.
- [ ] Test: same binary WITHOUT proxy — Qiling fails as it does today. No regression.
- [ ] Test: binary that calls `socket()` — existing Qiling handler runs (user did
NOT forward `socket`). Verify no interference.
- [ ] Test: user `set_syscall("epoll_create", my_hook)` — user hook takes priority
over proxy hook (user hook registered after proxy hook overwrites it via
`set_syscall`). Verify user control is preserved.
- [ ] Test: proxy process crash — verify Qiling reports error cleanly, doesn't hang.
**Existing code modified**: NONE. All new files in `qiling/os/posix/kernel_proxy/`.
Integration is purely through `set_syscall()`.
**Risk**: LOW — new code in new module. Existing behavior completely unchanged unless
user explicitly creates a `KernelProxy` and calls `forward_syscall()`.
---
## Phase 1: Networking Foundation
**Goal**: Forward all socket syscalls. Make a TCP client work end-to-end.
### 1.1 `ql_proxy_socket` FD type
New class in `qiling/os/posix/filestruct.py` (or new file alongside it). Must match
`ql_socket` interface so generic I/O dispatches correctly:
```python
class ql_proxy_socket:
"""Socket FD whose real socket lives in the kernel proxy process."""
def read(self, length: int) -> bytes:
# Forward to proxy: recv(self.proxy_fd, length)
def write(self, data: bytes) -> int:
# Forward to proxy: send(self.proxy_fd, data)
def close(self) -> None:
# Forward to proxy: close(self.proxy_fd)
def fileno(self) -> int:
# Return a sentinel — not a real host FD
# Socket-specific methods forwarded to proxy:
def connect(self, address) -> None: ...
def bind(self, address) -> None: ...
def listen(self, backlog) -> None: ...
def accept(self) -> tuple: ...
def shutdown(self, how) -> None: ...
def setsockopt(self, level, optname, value) -> None: ...
def getsockopt(self, level, optname) -> ...: ...
```
Because `ql_syscall_read` / `ql_syscall_write` / `ql_syscall_close` already dispatch
through `ql.os.fd[fd].read()` / `.write()` / `.close()`, these **existing handlers
need no changes** — the proxy socket object handles forwarding internally.
- [ ] Implement `ql_proxy_socket` class
- [ ] IPC client method for each operation
- [ ] Verify generic `read(fd, ...)` and `write(fd, ...)` work on proxy sockets
without modifying `ql_syscall_read` or `ql_syscall_write`
### 1.2 Socket syscall CALL hooks
Register CALL hooks for socket-specific syscalls. These are needed because socket
syscalls (bind, connect, listen, accept, etc.) have special argument handling
(sockaddr structs, address lengths) that goes beyond generic read/write.
- [ ] `socket()` — create proxy socket, store `ql_proxy_socket` in FD table
- [ ] `bind(fd, addr, addrlen)` — read sockaddr from guest memory, forward to proxy
- [ ] `connect(fd, addr, addrlen)` — same pattern
- [ ] `listen(fd, backlog)` — forward
- [ ] `accept(fd, addr, addrlen)` — forward, create new `ql_proxy_socket` for client FD
- [ ] `send/sendto/sendmsg` — read buffer from guest memory, forward
- [ ] `recv/recvfrom/recvmsg` — forward, write received data to guest memory
- [ ] `setsockopt/getsockopt` — forward with option translation
- [ ] `getpeername/getsockname` — forward, write sockaddr to guest memory
- [ ] `shutdown` — forward
- [ ] `socketpair` — forward, create two `ql_proxy_socket` objects
- [ ] `close` on proxy sockets — handled by `ql_proxy_socket.close()`,
but also register hook to detect close on proxy FDs if needed
**Struct translation**: sockaddr family (AF_INET, AF_INET6, AF_UNIX) is the same
across architectures. Network byte order is architecture-independent. The main
concern is pointer width (32-bit guest on 64-bit host) — read the right number
of bytes from guest memory based on `ql.arch.pointersize`.
### 1.3 `socketcall()` multiplexer (x86 32-bit)
On x86 32-bit, all socket operations go through a single `socketcall()` syscall
(`qiling/os/posix/syscall/net.py`). The existing multiplexer dispatches to individual
handlers. Since we hook the individual handlers (bind, connect, etc.), this works
automatically. But verify:
- [ ] Test that x86 32-bit socket operations are correctly forwarded via the
existing socketcall → individual handler → our CALL hook chain
### 1.4 Validation
- [ ] TCP client: connect to a server, send/receive data, close
- [ ] TCP server: bind, listen, accept, handle client, close
- [ ] UDP: sendto/recvfrom
- [ ] Unix domain sockets (path-based)
- [ ] Existing non-network tests still pass (regression check)
**Risk**: LOW-MEDIUM — new code + new FD type, but existing handlers and dispatch
untouched. Main risk is FD lifecycle bugs (leak, double-close).
---
## Phase 2: Complete Networking
**Goal**: epoll, network namespaces, advanced operations. Real-world network
binaries work.
### 2.1 epoll forwarding
epoll is currently **not implemented at all** — mapped in the syscall table but
no handler. This is new functionality, not a change to existing behavior.
- [ ] `epoll_create` / `epoll_create1` — forward, return proxy epoll FD
(new FD type or reuse `ql_proxy_socket` with a flag)
- [ ] `epoll_ctl(epfd, op, fd, event)` — forward; `fd` must be translated to
proxy FD space. Read `epoll_event` struct from guest memory.
- [ ] `epoll_wait(epfd, events, maxevents, timeout)` — forward. **This blocks**
in the proxy. For single-threaded programs this is correct (binary would be
blocked anyway). Write returned events to guest memory.
- [ ] `epoll_pwait` — same as epoll_wait + signal mask
**Blocking concern**: when the proxy is blocked on `epoll_wait`, the Unicorn
emulation is paused. This is correct for single-threaded programs. For multithreaded
programs, we need real threading (Phase 3) where each thread has its own Unicorn
and can block independently.
### 2.2 poll/select integration
Currently `poll()` and `select()` use host `select.poll()`/`select.select()` directly,
which won't work for proxy FDs (no host FD to poll).
- [ ] Hook `poll()` and `select()` — for FD sets containing proxy FDs, forward the
entire operation to the proxy
- [ ] For mixed FD sets (some proxy, some local): forward the proxy FDs to the proxy,
poll local FDs locally, merge results. This is complex — consider forwarding all
FDs to the proxy as the simpler approach.
### 2.3 Network namespace isolation
- [ ] Proxy process runs in its own network namespace (`unshare(CLONE_NEWNET)`)
- [ ] Configurable modes:
- `host`: proxy shares host network (default, simplest)
- `isolated`: separate namespace, no connectivity
- `bridged`: veth pair with NAT to host
- [ ] DNS: mount a resolv.conf in the proxy's mount namespace if needed
### 2.4 Advanced operations (lower priority)
- [ ] `sendmmsg` / `recvmmsg` — batch send/receive
- [ ] Raw sockets / packet sockets (`AF_PACKET`)
- [ ] Netlink sockets (`AF_NETLINK`) — for binaries that call `ip`, `route`, etc.
- [ ] `SCM_RIGHTS` (FD passing over Unix sockets) — requires FD translation
- [ ] IPv6 multicast
### 2.5 Validation
- [ ] epoll-based TCP echo server
- [ ] HTTP client (wget/curl-like binary)
- [ ] Binary that uses poll() with mixed file + socket FDs
- [ ] Network namespace: verify proxy and emulated binary are isolated from host
- [ ] Performance: measure latency overhead of IPC per syscall
**Risk**: MEDIUM — epoll is new functionality (no regression risk), but poll/select
changes for proxy FDs touch existing handlers. The mixed-FD-set case is the main
complexity.
---
## Phase 3: Real Threading
**Goal**: Real concurrency with one Unicorn engine per thread.
**This phase is high-risk and should only start after Phase 2 is stable.** It touches
the Unicorn integration, memory manager, thread lifecycle, and scheduler — all core
components. Needs a detailed design document before implementation begins.
### Prerequisites
- [ ] Phase 1-2 networking is stable and tested
- [ ] Detailed design document covering memory sharing, thread lifecycle,
and failure modes
- [ ] Prototype benchmark: measure overhead of multiple Unicorn instances
sharing memory via `mem_map_ptr`
### 3.1 Shared memory backing for Unicorn
Currently `QlMemoryManager.map()` calls `uc.mem_map()` which allocates internal
Unicorn memory. For shared threading, all Unicorn instances must see the same memory.
- [ ] Change memory backing to use `mmap(MAP_SHARED)` + `uc.mem_map_ptr()`
- [ ] This affects: `QlMemoryManager.map()`, `QlMemoryManager.protect()`,
`QlMemoryManager.unmap()`
- [ ] Loader changes: ELF/PE/MachO loaders must write segments into shared-backed
memory regions
- [ ] MMIO regions stay callback-based (not shared)
- [ ] **Critical**: This must be done as a standalone change that passes ALL existing
tests before moving to 3.2. If existing tests break, the shared memory
implementation is wrong.
Files: `qiling/os/memory.py`, `qiling/loader/elf.py`, `qiling/loader/pe.py`
### 3.2 One Unicorn instance per thread
Replace gevent Greenlets with real OS threads, each owning a Unicorn instance.
- [ ] New thread class: `QlLinuxRealThread` (alongside existing `QlLinuxThread`)
- Creates a new `Uc` instance on spawn
- Maps all shared memory regions into the new Uc via `mem_map_ptr`
- Copies parent registers to child Uc
- Sets child's SP, TLS, return value
- Runs in a real `threading.Thread`
- [ ] Modify `clone()` handler: when hybrid threading is enabled, create
`QlLinuxRealThread` instead of gevent Greenlet
- [ ] Per-thread hook context: each Unicorn instance needs its own hooks registered.
User-defined hooks must be replicated to all instances.
- [ ] Remove the 32337-instruction cooperative scheduling loop — real OS scheduler
handles preemption
**What breaks**: The current model assumes ONE `ql.uc` instance. Code that accesses
`ql.uc` directly will see only one thread's Unicorn. Need to audit all `ql.uc`
references and route to the current thread's instance.
Risky references:
- `ql.arch.regs` reads/writes `ql.uc` registers — must route to current thread's Uc
- `ql.mem.read/write` calls `ql.uc.mem_read/write` — with shared memory, any Uc works
- `ql.hook_*` registers on `ql.uc` — must register on all Uc instances
- `ql.save()/restore()` snapshots `ql.uc` — must snapshot correct thread
### 3.3 Synchronization primitives via real kernel
With real OS threads sharing real memory, kernel synchronization works natively.
- [ ] Forward `futex()` to kernel — `FUTEX_WAIT`/`FUTEX_WAKE` operate on the shared
memory addresses directly
- [ ] Remove gevent Event-based futex emulation (`qiling/os/linux/futex.py`)
- [ ] Forward `set_robust_list`, `get_robust_list`
- [ ] `pthread_mutex_*`, `pthread_cond_*` — these use futex internally, so forwarding
futex is sufficient
### 3.4 Thread safety for shared state
With real concurrent threads, shared mutable state needs synchronization.
- [ ] `QlMemoryManager`: lock `map_info` list mutations (map, unmap, protect)
- read/write don't need locks if backed by shared mmap (atomic at OS level)
- [ ] `QlFileDes`: lock FD table mutations (open, close, dup)
- [ ] Hook lists: lock registration/deregistration (hooks are usually set up before
`run()`, so contention should be minimal)
- [ ] Logging: thread-safe log handler with thread ID prefix
### 3.5 Validation
- [ ] pthread_create / pthread_join
- [ ] Mutex: two threads incrementing a shared counter with proper locking
- [ ] Condition variables: producer-consumer
- [ ] Futex: custom futex-based synchronization
- [ ] Thread-local storage (TLS) correctness per architecture
- [ ] Stress test: 10+ threads doing concurrent work
- [ ] ALL existing single-threaded tests still pass
- [ ] ALL existing gevent-threaded tests still pass (gevent mode preserved as fallback)
**Risk**: HIGH — changes to memory manager, Unicorn integration, and thread model.
Keep the existing gevent threading as a fallback mode. The new threading is opt-in.
---
## Phase 4: Signals
**Depends on Phase 3** (real threads required for proper signal delivery).
### 4.1 Signal handler registration
- [ ] Forward `sigaction(signum, act, oldact)` to kernel proxy
- [ ] Forward `sigprocmask` / `rt_sigprocmask`
- [ ] Forward `sigaltstack`
### 4.2 Signal delivery
When a signal is delivered to a proxy thread:
- [ ] Proxy catches the signal and sends notification to Qiling via IPC
- [ ] Qiling calls `emu_stop()` on the target thread's Unicorn
- [ ] Save thread context (registers)
- [ ] Build signal frame on emulated stack (architecture-specific)
- [ ] Set PC to the registered signal handler
- [ ] Resume Unicorn — handler executes in emulated code
- [ ] On `sigreturn` / `rt_sigreturn`: restore saved context, resume normal execution
### 4.3 Signal-syscall interaction
- [ ] `EINTR` on interrupted blocking syscalls
- [ ] `SA_RESTART` flag: automatically restart interrupted syscalls
- [ ] `kill()`, `tgkill()`, `tkill()` → forward to kernel
### 4.4 Validation
- [ ] SIGALRM handler (timer-based)
- [ ] SIGCHLD on child exit
- [ ] SIGPIPE on broken pipe
- [ ] Signal interrupting `read()` — verify EINTR
- [ ] Custom signal handler that modifies emulated state
**Risk**: MEDIUM — signal frame construction is architecture-specific and fiddly,
but the mechanism is well-understood. Main risk is getting the frame layout exactly
right for each architecture.
---
## Phase 5: Integration and Polish
### 5.1 User-facing API
```python
# Opt-in to hybrid kernel
ql = Qiling(argv=[...], rootfs="...")
# Enable kernel proxy for networking (Phase 1-2)
ql.os.kernel_proxy.enable(networking=True)
# Enable real threading (Phase 3) — requires networking=True
ql.os.kernel_proxy.enable(networking=True, threading=True)
# Enable signals (Phase 4) — requires threading=True
ql.os.kernel_proxy.enable(networking=True, threading=True, signals=True)
# Configure network namespace
ql.os.kernel_proxy.network_mode = "bridged" # "host" | "isolated" | "bridged"
# User hooks still work — they fire before/after proxy forwarding
ql.os.set_syscall("connect", my_connect_hook, QL_INTERCEPT.ENTER)
```
### 5.2 Backward compatibility
- [ ] Default behavior: no proxy, existing Python handlers — zero regression
- [ ] All existing tests pass with proxy disabled
- [ ] All existing tests pass with proxy enabled (forwarded syscalls should
produce equivalent results)
- [ ] `set_syscall()` user hooks fire correctly in both modes
- [ ] Existing gevent threading preserved as fallback when real threading not enabled
### 5.3 Fallback on failure
- [ ] If proxy process crashes: log error, fall back to Python handlers, warn user
- [ ] If proxy not available (non-Linux host): use Python handlers, warn user
- [ ] Graceful degradation: never crash, always fall back
### 5.4 Platform support
- [ ] Linux host: full support (namespaces, real threading)
- [ ] macOS host: proxy via Docker/Lima (networking only, no native namespaces)
- [ ] Windows host: proxy via WSL2 (networking only)
- [ ] Document host requirements
### 5.5 Performance
- [ ] Benchmark: syscall latency (Python handler vs proxy round-trip)
- [ ] Optimize IPC: shared memory ring buffer for high-frequency syscalls
- [ ] Batch small syscalls where possible
- [ ] Profile and tune for common workloads (network servers, threaded computation)
---
## Existing Issues (Independent of Hybrid Architecture)
These should be fixed regardless of the hybrid work.
### Bare except blocks swallowing errors
10+ bare `except:` blocks silently hide failures:
- `qiling/utils.py:242` — PE detection
- `qiling/debugger/qdb/qdb.py:128,352,598` — debugger operations
- `qiling/os/posix/filestruct.py:62,173,179` — fcntl/ioctl
- `qiling/os/posix/syscall/select.py:78` — select failures
- `qiling/os/windows/registry.py:127,185` — registry operations
### Asserts used for validation
Assertions disabled with `python -O`. Replace with exceptions:
- `qiling/os/memory.py` — page alignment, size, mapping checks
- `qiling/arch/x86_utils.py` — GDT/segment validation
- `qiling/cc/__init__.py` — calling convention validation
### Memory manager: string label parsing
`qiling/os/memory.py:209-218` — `get_lib_base()` uses regex on info strings.
Needs a proper mapping structure.
### ARM Thumb mode detection
`core.py:753` — fragile `_init_thumb` flag. Needs upstream Unicorn fix.
### x86 GDT privilege levels
`qiling/arch/x86_utils.py:147,178` — ring 3 forced to ring 0.
### Unbounded `read_cstring`
`qiling/os/memory.py:51-63` — no length limit. Can hang on MMIO.
### Incomplete save/restore
`QlOs.save()/restore()` empty in base class. UEFI and Windows don't implement it.
### Incomplete Windows emulation
Fiber, registry, handle management, DLL resolution gaps.
See `qiling/os/windows/` TODO comments.
### macOS and UEFI gaps
- macOS kext: 5 FIXMEs in `macos.py:79-117`
- UEFI variables: `uefi/rt.py:204-205`
### Hook system cleanup
- `type()` vs `isinstance()` in `core_hooks.py`
- Unclear return value semantics
- Non-intuitive `begin=1, end=0` for "entire memory"
### Hardcoded magic numbers
- Exit points in `os/os.py:84-87`
- Guard page `0x9000000` in `core.py:525`
### Test coverage
- ARM test skipped (`test_elf.py:411`)
- Multithread test skipped (`test_elf_multithread.py:185`)
- Broken wchar (`test_struct.py:170,185`)
- PowerPC, QNX, DOS, MCU: minimal coverage
+60
View File
@@ -2223,6 +2223,7 @@ riscv32_syscall_table = {
432: "fsmount",
433: "fspick",
434: "pidfd_open",
435: "clone3",
436: "close_range",
437: "openat2",
438: "pidfd_getfd",
@@ -2234,7 +2235,22 @@ riscv32_syscall_table = {
444: "landlock_create_ruleset",
445: "landlock_add_rule",
446: "landlock_restrict_self",
447: "memfd_secret",
448: "process_mrelease",
449: "futex_waitv",
450: "set_mempolicy_home_node",
451: "cachestat",
452: "fchmodat2",
453: "map_shadow_stack",
454: "futex_wake",
455: "futex_wait",
456: "futex_requeue",
457: "statmount",
458: "listmount",
459: "lsm_get_self_attr",
460: "lsm_set_self_attr",
461: "lsm_list_modules",
462: "mseal",
1024: "open",
1025: "link",
1026: "unlink",
@@ -2549,7 +2565,22 @@ riscv64_syscall_table = {
444: "landlock_create_ruleset",
445: "landlock_add_rule",
446: "landlock_restrict_self",
447: "memfd_secret",
448: "process_mrelease",
449: "futex_waitv",
450: "set_mempolicy_home_node",
451: "cachestat",
452: "fchmodat2",
453: "map_shadow_stack",
454: "futex_wake",
455: "futex_wait",
456: "futex_requeue",
457: "statmount",
458: "listmount",
459: "lsm_get_self_attr",
460: "lsm_set_self_attr",
461: "lsm_list_modules",
462: "mseal",
}
ppc_syscall_table = {
@@ -2968,4 +2999,33 @@ ppc_syscall_table = {
431: "fsconfig",
432: "fsmount",
433: "fspick",
434: "pidfd_open",
435: "clone3",
436: "close_range",
437: "openat2",
438: "pidfd_getfd",
439: "faccessat2",
440: "process_madvise",
441: "epoll_pwait2",
442: "mount_setattr",
443: "quotactl_fd",
444: "landlock_create_ruleset",
445: "landlock_add_rule",
446: "landlock_restrict_self",
447: "memfd_secret",
448: "process_mrelease",
449: "futex_waitv",
450: "set_mempolicy_home_node",
451: "cachestat",
452: "fchmodat2",
453: "map_shadow_stack",
454: "futex_wake",
455: "futex_wait",
456: "futex_requeue",
457: "statmount",
458: "listmount",
459: "lsm_get_self_attr",
460: "lsm_set_self_attr",
461: "lsm_list_modules",
462: "mseal",
}
+234
View File
@@ -0,0 +1,234 @@
#!/usr/bin/env python3
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
"""
Hybrid kernel proxy — forward specific syscalls to a real Linux kernel.
Usage:
from qiling import Qiling
from qiling.os.posix.kernel_proxy import KernelProxy
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")
ql.run()
"""
from __future__ import annotations
import os
import sys
import socket
import subprocess
import logging
from typing import Optional, TYPE_CHECKING
from qiling.const import QL_INTERCEPT, QL_OS
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.
The proxy process executes real syscalls and returns results. Integration
is through set_syscall() CALL hooks — no changes to Qiling's dispatch code.
"""
def __init__(self, ql: Qiling):
if sys.platform != 'linux':
raise RuntimeError("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._start_proxy()
def _start_proxy(self):
"""Start the proxy subprocess, connected via Unix socketpair."""
parent_sock, child_sock = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM)
child_fd = child_sock.fileno()
# ensure the subprocess can find qiling even when run from a subdirectory
env = os.environ.copy()
qiling_root = os.path.dirname(os.path.dirname(os.path.dirname(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))))))
python_path = env.get('PYTHONPATH', '')
env['PYTHONPATH'] = f"{qiling_root}:{python_path}" if python_path else qiling_root
self._process = subprocess.Popen(
[sys.executable, '-m', 'qiling.os.posix.kernel_proxy.proxy', str(child_fd)],
pass_fds=(child_fd,),
close_fds=True,
env=env,
)
child_sock.close()
self._client = ProxyClient(parent_sock)
log.info(f"kernel proxy started (pid={self._process.pid})")
def _build_reverse_table(self) -> dict:
"""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
arch_tables = {
QL_ARCH.ARM64 : 'arm64_syscall_table',
QL_ARCH.ARM : 'arm_syscall_table',
QL_ARCH.X8664 : 'x8664_syscall_table',
QL_ARCH.X86 : 'x86_syscall_table',
QL_ARCH.MIPS : 'mips_syscall_table',
QL_ARCH.RISCV : 'riscv32_syscall_table',
QL_ARCH.RISCV64 : 'riscv64_syscall_table',
QL_ARCH.PPC : 'ppc_syscall_table',
}
table_name = arch_tables.get(self.ql.arch.type)
if table_name is None:
raise RuntimeError(f"KernelProxy: unsupported architecture {self.ql.arch.type}")
import qiling.os.linux.map_syscall as mod
table = getattr(mod, table_name)
# reverse: name -> nr
self._reverse_table = {name: nr for nr, name in table.items()}
return self._reverse_table
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(
f"KernelProxy: syscall '{name}' not found in {self.ql.arch.type.name} syscall table"
)
return table[name]
def forward_syscall(self, name: str, returns_fd: bool = False):
"""Register a CALL hook that forwards this syscall to the kernel proxy.
Args:
name: syscall name (e.g. "epoll_create", "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.)
"""
nr = self._resolve_syscall_nr(name)
self._forwarded[name] = nr
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 ''}")
def _make_forwarder(self, name: str, guest_nr: int, returns_fd: bool):
"""Create a CALL hook closure for one syscall."""
client = self._client
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)
padded = args + (0,) * (6 - len(args))
retval = client.syscall(host_nr, padded[:6])
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)
ql.log.debug(f"kernel_proxy: {name}() -> proxy_fd={retval}, guest_fd={guest_fd}")
return guest_fd
ql.log.debug(f"kernel_proxy: {name}({', '.join(f'{a:#x}' for a in args)}) = {retval}")
return retval
_forwarder.__name__ = f'ql_syscall_{name}'
return _forwarder
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
if not hasattr(self, '_host_table'):
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")
return self._host_table[name]
def _load_host_syscall_table(self) -> dict:
"""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()
host_arch_map = {
'x86_64': 'x8664_syscall_table',
'aarch64': 'arm64_syscall_table',
'armv7l': 'arm_syscall_table',
'mips': 'mips_syscall_table',
'riscv64': 'riscv64_syscall_table',
'ppc': 'ppc_syscall_table',
}
table_name = host_arch_map.get(machine)
if table_name is None:
raise RuntimeError(f"KernelProxy: unsupported host architecture '{machine}'")
table = getattr(mod, table_name)
return {name: nr for nr, name in table.items()}
@staticmethod
def _alloc_fd(ql, fd_obj) -> int:
"""Find next free slot in Qiling's FD table and store fd_obj."""
for i in range(len(ql.os.fd)):
if ql.os.fd[i] is None:
ql.os.fd[i] = fd_obj
return i
raise OSError("kernel_proxy: FD table full")
def stop(self):
"""Stop the proxy process."""
if self._client:
try:
self._client.close()
except Exception:
pass
self._client = None
if self._process:
self._process.terminate()
try:
self._process.wait(timeout=5)
except subprocess.TimeoutExpired:
self._process.kill()
self._process.wait()
log.info(f"kernel proxy stopped (pid={self._process.pid})")
self._process = None
def __del__(self):
if hasattr(self, '_client'):
self.stop()
+176
View File
@@ -0,0 +1,176 @@
#!/usr/bin/env python3
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
"""
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)
All messages are length-prefixed binary over a Unix socketpair.
"""
import struct
import socket
from enum import IntEnum
class MsgType(IntEnum):
SYSCALL = 1
FD_OP = 2
class FdOp(IntEnum):
READ = 1
WRITE = 2
CLOSE = 3
DUP = 4
FCNTL = 5
IOCTL = 6
# 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]
#
# 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]
HEADER_FMT = '!BI' # msg_type/status (u8) + payload_len (u32)
HEADER_SIZE = struct.calcsize(HEADER_FMT)
SYSCALL_REQ_FMT = '!I6q' # syscall_nr (u32) + 6 args (i64)
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)
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)
FD_OP_RESP_FMT = '!qiI' # return_value (i64) + errno (i32) + data_len (u32)
FD_OP_RESP_SIZE = struct.calcsize(FD_OP_RESP_FMT)
def _recvall(sock: socket.socket, n: int) -> bytes:
"""Receive exactly n bytes from a socket."""
buf = bytearray()
while len(buf) < n:
chunk = sock.recv(n - len(buf))
if not chunk:
raise ConnectionError("kernel proxy connection closed")
buf.extend(chunk)
return bytes(buf)
class ProxyClient:
"""Qiling-side IPC client — sends requests to the proxy process."""
def __init__(self, sock: socket.socket):
self._sock = sock
def syscall(self, nr: int, args: tuple) -> int:
"""Forward a raw syscall. Returns kernel-convention result (negative errno on error)."""
padded = tuple(args) + (0,) * (6 - len(args))
payload = struct.pack(SYSCALL_REQ_FMT, nr, *padded[:6])
# send request
header = struct.pack(HEADER_FMT, MsgType.SYSCALL, len(payload))
self._sock.sendall(header + payload)
# recv response
resp_header = _recvall(self._sock, HEADER_SIZE)
_, resp_len = struct.unpack(HEADER_FMT, resp_header)
resp_payload = _recvall(self._sock, resp_len)
retval, errno_val = struct.unpack(SYSCALL_RESP_FMT, resp_payload)
return retval
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))
payload += data
header = struct.pack(HEADER_FMT, MsgType.FD_OP, len(payload))
self._sock.sendall(header + 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_val, data_len = struct.unpack(FD_OP_RESP_FMT, resp_payload[:FD_OP_RESP_SIZE])
resp_data = resp_payload[FD_OP_RESP_SIZE:FD_OP_RESP_SIZE + data_len]
return retval, resp_data
def fd_read(self, proxy_fd: int, length: int) -> bytes:
retval, data = self._fd_op(FdOp.READ, proxy_fd, arg1=length)
if retval < 0:
return b''
return data
def fd_write(self, proxy_fd: int, data: bytes) -> int:
retval, _ = self._fd_op(FdOp.WRITE, proxy_fd, data=data)
return retval
def fd_close(self, proxy_fd: int) -> None:
self._fd_op(FdOp.CLOSE, proxy_fd)
def fd_dup(self, proxy_fd: int) -> int:
retval, _ = self._fd_op(FdOp.DUP, proxy_fd)
return retval
def fd_fcntl(self, proxy_fd: int, cmd: int, arg: int) -> int:
retval, _ = self._fd_op(FdOp.FCNTL, proxy_fd, arg1=cmd, arg2=arg)
return retval
def fd_ioctl(self, proxy_fd: int, cmd: int, arg: int) -> int:
retval, _ = self._fd_op(FdOp.IOCTL, proxy_fd, arg1=cmd, arg2=arg)
return retval
def close(self):
self._sock.close()
class ProxyServer:
"""Proxy-side IPC server — receives requests, executes real syscalls."""
def __init__(self, sock: socket.socket):
self._sock = sock
def recv_request(self) -> tuple:
"""Receive one request. Returns (msg_type, parsed_fields)."""
header = _recvall(self._sock, HEADER_SIZE)
msg_type, payload_len = struct.unpack(HEADER_FMT, header)
payload = _recvall(self._sock, payload_len)
if msg_type == MsgType.SYSCALL:
fields = struct.unpack(SYSCALL_REQ_FMT, payload)
return MsgType.SYSCALL, fields # (nr, a0, a1, a2, a3, a4, a5)
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
data = payload[FD_OP_REQ_SIZE:FD_OP_REQ_SIZE + data_len]
return MsgType.FD_OP, (FdOp(op), proxy_fd, arg1, arg2, data)
else:
raise ValueError(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)
header = struct.pack(HEADER_FMT, 0, len(payload))
self._sock.sendall(header + 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
header = struct.pack(HEADER_FMT, 0, len(payload))
self._sock.sendall(header + payload)
+116
View File
@@ -0,0 +1,116 @@
#!/usr/bin/env python3
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
"""
Kernel proxy process — executes real Linux syscalls on behalf of Qiling.
This runs as a subprocess. It receives syscall requests over a Unix socketpair,
executes them via libc.syscall(), and sends back results.
Usage (internal — started by KernelProxy.__init__):
python -m qiling.os.posix.kernel_proxy.proxy <socket_fd>
"""
import os
import sys
import ctypes
import ctypes.util
import errno as errno_module
import socket
from qiling.os.posix.kernel_proxy.ipc import (
ProxyServer, MsgType, FdOp
)
# 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)
sys.exit(1)
_libc = ctypes.CDLL(_libc_path, use_errno=True)
_libc.syscall.restype = ctypes.c_long
_libc.syscall.argtypes = [ctypes.c_long] + [ctypes.c_long] * 6
def raw_syscall(nr: int, a0: int, a1: int, a2: int, a3: int, a4: int, a5: int) -> tuple:
"""Execute a real Linux syscall. Returns (return_value, errno) in kernel convention."""
ctypes.set_errno(0)
result = _libc.syscall(nr, a0, a1, a2, a3, a4, a5)
if result == -1:
err = ctypes.get_errno()
if err != 0:
return -err, err # kernel convention: negative errno
return result, 0
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:
if op == FdOp.READ:
result = os.read(proxy_fd, arg1)
return len(result), 0, result
elif op == FdOp.WRITE:
written = os.write(proxy_fd, data)
return written, 0, b''
elif op == FdOp.CLOSE:
os.close(proxy_fd)
return 0, 0, b''
elif op == FdOp.DUP:
new_fd = os.dup(proxy_fd)
return new_fd, 0, b''
elif op == FdOp.FCNTL:
import fcntl
result = fcntl.fcntl(proxy_fd, arg1, arg2)
return result, 0, b''
elif op == FdOp.IOCTL:
import fcntl
result = fcntl.ioctl(proxy_fd, arg1, arg2)
return result, 0, b''
else:
return -errno_module.ENOSYS, errno_module.ENOSYS, b''
except OSError as e:
return -e.errno, e.errno, b''
def main():
if len(sys.argv) != 2:
print(f"usage: {sys.argv[0]} <socket_fd>", file=sys.stderr)
sys.exit(1)
sock_fd = int(sys.argv[1])
sock = socket.socket(fileno=sock_fd)
server = ProxyServer(sock)
while True:
try:
msg_type, fields = server.recv_request()
except ConnectionError:
break
if msg_type == MsgType.SYSCALL:
nr, a0, a1, a2, a3, a4, a5 = fields
retval, err = raw_syscall(nr, a0, a1, a2, a3, a4, a5)
server.send_syscall_response(retval, err)
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)
server.send_fd_op_response(retval, err, resp_data)
sock.close()
if __name__ == '__main__':
main()
+47
View File
@@ -0,0 +1,47 @@
#!/usr/bin/env python3
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
"""
Proxy file descriptor — wraps an FD that lives in the kernel proxy process.
When a forwarded syscall returns an FD (e.g. epoll_create, eventfd), the real
FD lives in the proxy. This wrapper forwards read/write/close to the proxy via
IPC, matching the interface of ql_socket and ql_pipe so existing syscall handlers
(ql_syscall_read, ql_syscall_write, ql_syscall_close) work without modification.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from qiling.os.posix.kernel_proxy.ipc import ProxyClient
class ql_proxy_fd:
def __init__(self, client: ProxyClient, proxy_fd: int):
self._client = client
self._proxy_fd = proxy_fd
def read(self, length: int) -> bytes:
return self._client.fd_read(self._proxy_fd, length)
def write(self, data: bytes) -> int:
return self._client.fd_write(self._proxy_fd, data)
def close(self) -> None:
self._client.fd_close(self._proxy_fd)
def fileno(self) -> int:
return -1
def dup(self) -> ql_proxy_fd:
new_proxy_fd = self._client.fd_dup(self._proxy_fd)
return ql_proxy_fd(self._client, new_proxy_fd)
def fcntl(self, cmd, arg):
return self._client.fd_fcntl(self._proxy_fd, cmd, arg)
def ioctl(self, cmd, arg):
return self._client.fd_ioctl(self._proxy_fd, cmd, arg)
+355
View File
@@ -0,0 +1,355 @@
#!/usr/bin/env python3
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
import sys
import struct
import unittest
import platform
sys.path.append("..")
from qiling import Qiling
from qiling.const import QL_INTERCEPT, QL_VERBOSE
@unittest.skipUnless(platform.system() == 'Linux', 'kernel proxy requires Linux host')
class KernelProxyTest(unittest.TestCase):
"""Tests for the hybrid kernel proxy (Phase 0)."""
ROOTFS = "../examples/rootfs/x8664_linux"
HELLO_BIN = "../examples/rootfs/x8664_linux/bin/x8664_hello"
# -------------------------------------------------------------------------
# Proxy lifecycle
# -------------------------------------------------------------------------
def test_proxy_start_stop(self):
"""Proxy starts and stops cleanly."""
from qiling.os.posix.kernel_proxy import KernelProxy
ql = Qiling([self.HELLO_BIN], self.ROOTFS, verbose=QL_VERBOSE.OFF)
proxy = KernelProxy(ql)
self.assertIsNotNone(proxy._process)
self.assertIsNotNone(proxy._client)
self.assertTrue(proxy._process.poll() is None) # still running
proxy.stop()
self.assertIsNone(proxy._process)
self.assertIsNone(proxy._client)
del ql
def test_no_proxy_no_change(self):
"""Without proxy, Qiling behaves identically."""
from qiling.extensions import pipe
ql = Qiling([self.HELLO_BIN], self.ROOTFS, verbose=QL_VERBOSE.OFF)
ql.os.stdout = pipe.SimpleOutStream(1)
ql.run()
self.assertIn(b"Hello", ql.os.stdout.read(1024))
del ql
def test_proxy_attached_no_forwarding(self):
"""Proxy attached but no syscalls forwarded — no behavior change."""
from qiling.os.posix.kernel_proxy import KernelProxy
from qiling.extensions import pipe
ql = Qiling([self.HELLO_BIN], self.ROOTFS, verbose=QL_VERBOSE.OFF)
ql.os.stdout = pipe.SimpleOutStream(1)
proxy = KernelProxy(ql)
ql.run()
proxy.stop()
self.assertIn(b"Hello", ql.os.stdout.read(1024))
del ql
# -------------------------------------------------------------------------
# Raw syscall forwarding (integer-only args)
# -------------------------------------------------------------------------
def test_forward_getpid(self):
"""Forward getpid — proxy returns its own PID."""
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')
results = []
def on_getpid_exit(ql, *args):
results.append(args[-1])
ql.os.set_syscall('getpid', on_getpid_exit, QL_INTERCEPT.EXIT)
ql.run()
# if getpid was called, it should return the proxy's PID
if results:
self.assertEqual(results[0], proxy._process.pid)
proxy.stop()
del ql
def test_forward_brk(self):
"""Forward brk to real kernel — binary still runs correctly."""
from qiling.os.posix.kernel_proxy import KernelProxy
from qiling.extensions import pipe
ql = Qiling([self.HELLO_BIN], self.ROOTFS, verbose=QL_VERBOSE.OFF)
ql.os.stdout = pipe.SimpleOutStream(1)
proxy = KernelProxy(ql)
proxy.forward_syscall('brk')
brk_results = []
def on_brk_exit(ql, *args):
brk_results.append(args[-1])
ql.os.set_syscall('brk', on_brk_exit, QL_INTERCEPT.EXIT)
ql.run()
proxy.stop()
# brk is called during libc init
self.assertGreater(len(brk_results), 0, "brk was never called")
# binary should still produce correct output
self.assertIn(b"Hello", ql.os.stdout.read(1024))
del ql
# -------------------------------------------------------------------------
# FD-returning syscalls (returns_fd=True)
# -------------------------------------------------------------------------
def test_forward_returns_fd(self):
"""forward_syscall with returns_fd=True creates ql_proxy_fd in FD table."""
from qiling.os.posix.kernel_proxy import KernelProxy
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('eventfd2', returns_fd=True)
# call the forwarder directly (simulates binary calling eventfd2)
hook = ql.os.posix_syscall_hooks[QL_INTERCEPT.CALL].get('ql_syscall_eventfd2')
self.assertIsNotNone(hook, "forwarder hook not registered")
guest_fd = hook(ql, 0, 0)
self.assertGreaterEqual(guest_fd, 0)
fd_obj = ql.os.fd[guest_fd]
self.assertIsInstance(fd_obj, ql_proxy_fd)
# write and read through the proxy FD
fd_obj.write(struct.pack('<Q', 42))
data = fd_obj.read(8)
self.assertEqual(struct.unpack('<Q', data)[0], 42)
fd_obj.close()
proxy.stop()
del ql
def test_proxy_fd_dup(self):
"""ql_proxy_fd.dup() creates a working copy."""
from qiling.os.posix.kernel_proxy import KernelProxy
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('eventfd2', returns_fd=True)
hook = ql.os.posix_syscall_hooks[QL_INTERCEPT.CALL]['ql_syscall_eventfd2']
guest_fd = hook(ql, 0, 0)
fd_obj = ql.os.fd[guest_fd]
duped = fd_obj.dup()
self.assertIsInstance(duped, ql_proxy_fd)
self.assertNotEqual(fd_obj._proxy_fd, duped._proxy_fd)
# write on original, read on dup
fd_obj.write(struct.pack('<Q', 7))
data = duped.read(8)
self.assertEqual(struct.unpack('<Q', data)[0], 7)
fd_obj.close()
duped.close()
proxy.stop()
del ql
# -------------------------------------------------------------------------
# User hooks coexist with proxy hooks
# -------------------------------------------------------------------------
def test_user_exit_hook_fires_on_forwarded_syscall(self):
"""User EXIT hook fires after proxy CALL hook."""
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('brk')
exit_hook_called = []
def on_exit(ql, *args):
exit_hook_called.append(True)
ql.os.set_syscall('brk', on_exit, QL_INTERCEPT.EXIT)
ql.run()
proxy.stop()
self.assertGreater(len(exit_hook_called), 0,
"EXIT hook never fired on forwarded syscall")
del ql
def test_user_enter_hook_fires_on_forwarded_syscall(self):
"""User ENTER hook fires before proxy CALL hook."""
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('brk')
enter_hook_called = []
def on_enter(ql, *args):
enter_hook_called.append(args)
return None # don't override args
ql.os.set_syscall('brk', on_enter, QL_INTERCEPT.ENTER)
ql.run()
proxy.stop()
self.assertGreater(len(enter_hook_called), 0,
"ENTER hook never fired on forwarded syscall")
del ql
def test_user_call_hook_overrides_proxy(self):
"""User CALL hook registered after proxy takes priority."""
from qiling.os.posix.kernel_proxy import KernelProxy
from qiling.extensions import pipe
ql = Qiling([self.HELLO_BIN], self.ROOTFS, verbose=QL_VERBOSE.OFF)
ql.os.stdout = pipe.SimpleOutStream(1)
proxy = KernelProxy(ql)
proxy.forward_syscall('brk')
custom_brk_called = []
def my_brk(ql, addr):
custom_brk_called.append(addr)
# return a fake address
return 0x700000
# register user CALL hook AFTER proxy — should override
ql.os.set_syscall('brk', my_brk, QL_INTERCEPT.CALL)
ql.run()
proxy.stop()
self.assertGreater(len(custom_brk_called), 0,
"user CALL hook should override proxy hook")
del ql
# -------------------------------------------------------------------------
# IPC protocol
# -------------------------------------------------------------------------
def test_ipc_roundtrip(self):
"""IPC serialization/deserialization is correct."""
from qiling.os.posix.kernel_proxy.ipc import (
HEADER_FMT, HEADER_SIZE,
SYSCALL_REQ_FMT, SYSCALL_RESP_FMT,
FD_OP_REQ_FMT, FD_OP_RESP_FMT,
MsgType, FdOp
)
import struct
# syscall request roundtrip
payload = struct.pack(SYSCALL_REQ_FMT, 39, 1, 2, 3, 4, 5, 6)
nr, a0, a1, a2, a3, a4, a5 = struct.unpack(SYSCALL_REQ_FMT, payload)
self.assertEqual(nr, 39)
self.assertEqual((a0, a1, a2, a3, a4, a5), (1, 2, 3, 4, 5, 6))
# syscall response roundtrip
payload = struct.pack(SYSCALL_RESP_FMT, -2, 2) # -ENOENT
retval, errno_val = struct.unpack(SYSCALL_RESP_FMT, payload)
self.assertEqual(retval, -2)
self.assertEqual(errno_val, 2)
# -------------------------------------------------------------------------
# Error handling
# -------------------------------------------------------------------------
def test_forward_invalid_syscall_name(self):
"""forward_syscall with bogus name raises ValueError."""
from qiling.os.posix.kernel_proxy import KernelProxy
ql = Qiling([self.HELLO_BIN], self.ROOTFS, verbose=QL_VERBOSE.OFF)
proxy = KernelProxy(ql)
with self.assertRaises(ValueError):
proxy.forward_syscall('nonexistent_syscall_xyz')
proxy.stop()
del ql
def test_forward_syscall_error_returns_negative_errno(self):
"""Forwarded syscall that fails returns negative errno."""
from qiling.os.posix.kernel_proxy import KernelProxy
ql = Qiling([self.HELLO_BIN], self.ROOTFS, verbose=QL_VERBOSE.OFF)
proxy = KernelProxy(ql)
# directly test via the IPC client: close an invalid FD
retval = proxy._client.syscall(
proxy._get_host_syscall_nr('close'),
(9999, 0, 0, 0, 0, 0)
)
# should return -EBADF (-9)
self.assertEqual(retval, -9)
proxy.stop()
del ql
# -------------------------------------------------------------------------
# Syscall table resolution
# -------------------------------------------------------------------------
def test_host_table_loaded(self):
"""Host syscall table resolves common syscalls."""
from qiling.os.posix.kernel_proxy import KernelProxy
ql = Qiling([self.HELLO_BIN], self.ROOTFS, verbose=QL_VERBOSE.OFF)
proxy = KernelProxy(ql)
for name in ['read', 'write', 'close', 'socket', 'epoll_create1']:
nr = proxy._get_host_syscall_nr(name)
self.assertIsInstance(nr, int)
self.assertGreaterEqual(nr, 0)
proxy.stop()
del ql
def test_guest_table_reverse_lookup(self):
"""Guest syscall name resolves to correct number."""
from qiling.os.posix.kernel_proxy import KernelProxy
ql = Qiling([self.HELLO_BIN], self.ROOTFS, verbose=QL_VERBOSE.OFF)
proxy = KernelProxy(ql)
# x8664: write is syscall 1
nr = proxy._resolve_syscall_nr('write')
self.assertEqual(nr, 1)
# x8664: epoll_create1 is syscall 291
nr = proxy._resolve_syscall_nr('epoll_create1')
self.assertEqual(nr, 291)
proxy.stop()
del ql
if __name__ == "__main__":
unittest.main()