Hijack API

Hijack API

Standard Stream Hijacking

Replace stdin, stdout, or stderr with custom objects.

Feed static input to stdin

1from qiling import Qiling
2from qiling.extensions import pipe
3
4ql = Qiling([r'rootfs/x86_windows/bin/Easy_CrackMe.exe'], r'rootfs/x86_windows')
5ql.os.stdin = pipe.SimpleInStream(0)
6ql.os.stdin.write(b'Ea5yR3versing\n')
7ql.run()

Interactive stdin (pwntools-style)

1from qiling.extensions import pipe
2
3ql.os.stdin = pipe.InteractiveInStream()  # prompts user for input at runtime

VFS Object Mapping

Map virtual paths to host files or custom objects. Useful for procfs, sysfs, udev, or custom device simulation.

Map to host file

1ql.add_fs_mapper(r'/dev/urandom', r'/dev/urandom')

Map to custom object

Extend QlFsMappedObject and implement read, fstat, close:

 1from qiling.os.mapper import QlFsMappedObject
 2
 3class FakeUrandom(QlFsMappedObject):
 4    def read(self, size: int) -> bytes:
 5        return b"\x04"  # always return constant
 6    def fstat(self) -> int:
 7        return -1       # let syscall fstat ignore it
 8    def close(self) -> int:
 9        return 0
10
11ql.add_fs_mapper(r'/dev/urandom', FakeUrandom())

Disk emulation

1from qiling.os.disk import QlDisk
2
3emu_disk = QlDisk(r'rootfs/8086_dos/petya/out_1M.raw', 0x80)
4ql.add_fs_mapper(0x80, emu_disk)

QlDisk implements cylinder/head/sector and LBA logic. Drive index 0x80 = first hard disk in BIOS/DOS; Linux uses /dev/sda; Windows uses \\.\PHYSICALDRIVE0.

POSIX Syscall Interception

Intercept stages via QL_INTERCEPT:

 1from qiling import Qiling
 2from qiling.const import QL_INTERCEPT
 3
 4def my_syscall_write(ql: Qiling, fd: int, buf: int, count: int) -> int:
 5    data = ql.mem.read(buf, count)
 6    fobj = ql.os.fd[fd]
 7    if hasattr(fobj, 'write'):
 8        fobj.write(data)
 9    ret = count
10    ql.log.info(f'my_syscall_write({fd}, {buf:#x}, {count}) = {ret}')
11    return ret
12
13ql = Qiling([r'rootfs/arm_linux/bin/arm_hello'], r'rootfs/arm_linux')
14ql.os.set_syscall('write', my_syscall_write, QL_INTERCEPT.CALL)
15# Equivalent by syscall number:
16# ql.os.set_syscall(4, my_syscall_write)
17ql.run()

POSIX OS API Hooking

 1from qiling.const import QL_INTERCEPT
 2from qiling.os.const import STRING
 3
 4def my_puts(ql: Qiling):
 5    # Resolve call parameters by name and type
 6    params = ql.os.resolve_fcall_params({'s': STRING})
 7    s = params['s']
 8    ql.log.info(f'my_puts: got "{s}"')
 9    print(s)
10    return len(s)
11
12ql.os.set_api('puts', my_puts, QL_INTERCEPT.CALL)

Windows API Hooking

Use @winsdkapi decorator with calling convention and parameter types.

 1from qiling.os.windows.api import *
 2from qiling.os.windows.fncc import *
 3
 4@winsdkapi(cc=CDECL, params={
 5    'dest'  : POINTER,
 6    'src'   : POINTER,
 7    'count' : UINT
 8})
 9def my_memcpy(ql: Qiling, address: int, params):
10    data = bytes(ql.mem.read(params['src'], params['count']))
11    ql.mem.write(params['dest'], data)
12    return params['dest']

Note: cc is ignored on 64-bit (always treated as MS64 for compatibility).

Hook intercept stages for Windows APIs:

Memory leak detection example

 1chunks = set()
 2
 3@winsdkapi(cc=CDECL, params={'size': UINT})
 4def on_malloc_exit(ql, address, params, retval: int):
 5    chunks.add(retval)
 6
 7@winsdkapi(cc=CDECL, params={'address': POINTER})
 8def on_free_entry(ql, address, params):
 9    memaddr = params['address']
10    try:
11        chunks.remove(memaddr)
12    except KeyError:
13        ql.log.warning(f'possible double-free of {memaddr:#010x}')
14        params['address'] = 0
15        return address, params  # override to prevent crash
16
17ql.os.set_api("malloc", on_malloc_exit, QL_INTERCEPT.EXIT)
18ql.os.set_api("free", on_free_entry, QL_INTERCEPT.ENTER)

UEFI API Hooking

Use @dxeapi decorator (applies to both DXE and SMM):

 1from qiling.os.uefi.const import EFI_SUCCESS
 2from qiling.os.uefi.fncc import *
 3from qiling.os.uefi.ProcessorBind import *
 4
 5@dxeapi(params={
 6    "VariableName" : WSTRING,
 7    "VendorGuid"   : GUID,
 8    "Attributes"   : UINT,
 9    "DataSize"     : UINT,
10    "Data"         : POINTER
11})
12def hook_SetVariable(ql: Qiling, address: int, params):
13    data = ql.mem.read(params['Data'], params['DataSize'])
14    ql.env[params['VariableName']] = bytes(data)
15    return EFI_SUCCESS