mirror of
https://github.com/qilingframework/qiling.git
synced 2026-10-12 02:21:40 +00:00
retrocpugeekandClaude Opus 4.8
0e52738d53
Drives ql_syscall_statx on a big-endian MIPS context and checks that stx_mode, read back in the guest's byte order, still reports a directory (S_IFDIR). Before the fix the statx struct was emitted little-endian regardless of guest endianness, so the type bits were byte-swapped and a directory looked like a regular file. Self-contained (no rootfs binary, runs on stock unicorn). Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
979 lines
36 KiB
Python
979 lines
36 KiB
Python
#!/usr/bin/env python3
|
|
#
|
|
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
|
|
#
|
|
|
|
import unittest
|
|
import string
|
|
import random
|
|
import os
|
|
import io
|
|
import re
|
|
import socket
|
|
import sys
|
|
import time
|
|
import threading
|
|
from ctypes import *
|
|
sys.path.append("..")
|
|
|
|
from typing import Any, Sequence
|
|
|
|
from qiling import Qiling
|
|
from qiling.const import QL_ARCH, QL_OS, QL_INTERCEPT, QL_STOP, QL_VERBOSE
|
|
from qiling.exception import *
|
|
from qiling.extensions import pipe
|
|
from qiling.os.const import STRING
|
|
from qiling.os.posix import syscall
|
|
from qiling.os.mapper import QlFsMappedObject
|
|
|
|
|
|
class ELFTest(unittest.TestCase):
|
|
|
|
def test_libpatch_elf_linux_x8664(self):
|
|
ql = Qiling(["../examples/rootfs/x8664_linux/bin/patch_test.bin"], "../examples/rootfs/x8664_linux")
|
|
ql.patch(0x0000000000000575, b'qiling\x00', target='libpatch_test.so')
|
|
ql.run()
|
|
del ql
|
|
|
|
def test_elf_freebsd_x8664(self):
|
|
ql = Qiling(["../examples/rootfs/x8664_freebsd/bin/x8664_hello_asm"], "../examples/rootfs/x8664_freebsd", verbose=QL_VERBOSE.DUMP)
|
|
ql.run()
|
|
del ql
|
|
|
|
def test_elf_partial_linux_x8664(self):
|
|
snapshot_file = r'/tmp/snapshot.bin'
|
|
|
|
def dump(ql: Qiling, *args, **kw):
|
|
ql.save(reg=False, cpu_context=True, snapshot=snapshot_file)
|
|
ql.emu_stop()
|
|
|
|
ql = Qiling(["../examples/rootfs/x8664_linux/bin/sleep_hello"], "../examples/rootfs/x8664_linux", verbose=QL_VERBOSE.DEFAULT)
|
|
load_address = ql.profile.getint("OS64", "load_address")
|
|
ql.hook_address(dump, load_address + 0x1094)
|
|
ql.run()
|
|
|
|
ql = Qiling(["../examples/rootfs/x8664_linux/bin/sleep_hello"], "../examples/rootfs/x8664_linux", verbose=QL_VERBOSE.DEBUG)
|
|
load_address = ql.profile.getint("OS64", "load_address")
|
|
ql.restore(snapshot=snapshot_file)
|
|
|
|
begin_point = load_address + 0x109e
|
|
end_point = load_address + 0x10bc
|
|
|
|
ql.run(begin_point, end_point)
|
|
|
|
del ql
|
|
|
|
def test_elf_x_only_segment(self):
|
|
def stop(ql: Qiling):
|
|
ql.emu_stop()
|
|
|
|
ql = Qiling(["../examples/rootfs/x8664_linux/bin/sleep_hello_with_x_only_segment"], "../examples/rootfs/x8664_linux", verbose=QL_VERBOSE.DEBUG)
|
|
load_address = ql.profile.getint("OS64", "load_address")
|
|
ql.hook_address(stop, load_address + 0x1094)
|
|
ql.run()
|
|
|
|
del ql
|
|
|
|
def _test_elf_linux_x86_snapshot_restore_common(self, reg=False, ctx=False):
|
|
rootfs = "../examples/rootfs/x86_linux"
|
|
cmdline = ["../examples/rootfs/x86_linux/bin/x86_hello"]
|
|
snapshot = os.path.join(rootfs, 'snapshot_restore_reg_ctx.snapshot')
|
|
|
|
ql = Qiling(cmdline, rootfs, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
load_address = ql.profile.getint("OS32", "load_address")
|
|
hook_address = load_address + 0x542 # call printf
|
|
|
|
def dump(ql: Qiling):
|
|
ql.save(reg=reg, cpu_context=ctx, os=True, loader=True, snapshot=snapshot)
|
|
ql.emu_stop()
|
|
|
|
ql.hook_address(dump, hook_address)
|
|
ql.run()
|
|
|
|
# make sure that the ending PC is the same as the hook address because dump stops the emulater
|
|
self.assertEqual(ql.arch.regs.arch_pc, hook_address)
|
|
del ql
|
|
|
|
ql = Qiling(cmdline, rootfs, verbose=QL_VERBOSE.DEBUG)
|
|
ql.restore(snapshot=snapshot)
|
|
|
|
# ensure that the starting PC is same as the PC we stopped on when taking the snapshot
|
|
self.assertEqual(ql.arch.regs.arch_pc, hook_address)
|
|
|
|
ql.run(begin=hook_address)
|
|
del ql
|
|
|
|
os.remove(snapshot)
|
|
|
|
def test_elf_linux_x86_snapshot_restore_reg(self):
|
|
self._test_elf_linux_x86_snapshot_restore_common(reg=True, ctx=False)
|
|
|
|
def test_elf_linux_x86_snapshot_restore_ctx(self):
|
|
self._test_elf_linux_x86_snapshot_restore_common(reg=False, ctx=True)
|
|
|
|
def test_elf_linux_x86_snapshot_restore_reg_ctx(self):
|
|
self._test_elf_linux_x86_snapshot_restore_common(reg=True, ctx=True)
|
|
|
|
PARAMS_PUTS = {'s': STRING}
|
|
|
|
def test_elf_linux_x8664(self):
|
|
checklist = {}
|
|
|
|
def my_puts(ql: Qiling):
|
|
params = ql.os.resolve_fcall_params(ELFTest.PARAMS_PUTS)
|
|
print(f'puts("{params["s"]}")')
|
|
|
|
reg = ql.arch.regs.rax
|
|
print(f'reg : {reg:#x}')
|
|
|
|
checklist['set_api'] = reg
|
|
|
|
def write_onEnter(ql: Qiling, fd: int, str_ptr: int, str_len: int):
|
|
checklist['set_syscall_onenter'] = True
|
|
print("enter write syscall!")
|
|
|
|
# override syscall pc (ignored) and set of params with our own
|
|
return None, (fd, str_ptr + 1, str_len - 1)
|
|
|
|
def write_onexit(ql: Qiling, fd: int, str_ptr: int, str_len: int, retval: int):
|
|
checklist['set_syscall_onexit'] = True
|
|
print("exit write syscall!")
|
|
|
|
# override syscall return value with our own
|
|
return str_len + 1
|
|
|
|
ql = Qiling(["../examples/rootfs/x8664_linux/bin/x8664_args", "1234test", "12345678", "bin/x8664_hello"], "../examples/rootfs/x8664_linux", verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.set_syscall(1, write_onEnter, QL_INTERCEPT.ENTER)
|
|
ql.os.set_syscall(1, write_onexit, QL_INTERCEPT.EXIT)
|
|
ql.os.set_api('puts', my_puts)
|
|
|
|
blob = bytes.fromhex("ff fe fd fc fb fa fb fc fc fe fd")
|
|
|
|
ql.mem.map(0x1000, 0x1000)
|
|
ql.mem.write(0x1000, blob)
|
|
|
|
ql.mem.map(0x2000, 0x1000)
|
|
ql.mem.write(0x2000, blob)
|
|
|
|
ql.run()
|
|
|
|
self.assertListEqual([0x1000, 0x2000], ql.mem.search(blob))
|
|
self.assertEqual(0x5555555546ca, checklist['set_api'])
|
|
self.assertTrue(checklist['set_syscall_onenter'])
|
|
self.assertTrue(checklist['set_syscall_onexit'])
|
|
|
|
del ql
|
|
|
|
def test_elf_hijackapi_linux_x8664(self):
|
|
checklist = {}
|
|
|
|
def my_puts_enter(ql: Qiling):
|
|
params = ql.os.resolve_fcall_params(ELFTest.PARAMS_PUTS)
|
|
checklist['enter_str'] = params["s"]
|
|
|
|
def my_puts_exit(ql):
|
|
checklist['exit_rdi'] = ql.arch.regs.rdi
|
|
|
|
ql = Qiling(["../examples/rootfs/x8664_linux/bin/x8664_puts"], "../examples/rootfs/x8664_linux", verbose=QL_VERBOSE.DEBUG)
|
|
ql.os.set_api('puts', my_puts_enter, QL_INTERCEPT.ENTER)
|
|
ql.os.set_api('puts', my_puts_exit, QL_INTERCEPT.EXIT)
|
|
|
|
ql.run()
|
|
|
|
self.assertIn(checklist['exit_rdi'], (0x1, 0x7fffb81c2760))
|
|
self.assertEqual("CCCC", checklist['enter_str'])
|
|
|
|
del ql
|
|
|
|
def test_elf_linux_x8664_flex_api(self):
|
|
opened = []
|
|
|
|
def onenter_fopen(ql: Qiling):
|
|
params = ql.os.resolve_fcall_params({
|
|
'filename' : STRING,
|
|
'mode' : STRING
|
|
})
|
|
|
|
# log opened filenames
|
|
opened.append(params['filename'])
|
|
|
|
def hook_main(ql: Qiling):
|
|
# set up fopen hook when reaching main
|
|
ql.os.set_api('fopen', onenter_fopen, QL_INTERCEPT.ENTER)
|
|
|
|
ql = Qiling(["../examples/rootfs/x8664_linux/bin/x8664_fetch_urandom"], "../examples/rootfs/x8664_linux", verbose=QL_VERBOSE.DEFAULT)
|
|
|
|
ba = ql.loader.images[0].base
|
|
ql.hook_address(hook_main, ba + 0x10e0)
|
|
ql.run()
|
|
del ql
|
|
|
|
# test whether we interpected opening urandom
|
|
self.assertListEqual(opened, [r'/dev/urandom'])
|
|
|
|
def test_elf_linux_x8664_static(self):
|
|
ql = Qiling(["../examples/rootfs/x8664_linux/bin/x8664_hello_static"], "../examples/rootfs/x8664_linux", verbose=QL_VERBOSE.DEBUG)
|
|
ql.run()
|
|
del ql
|
|
|
|
def test_elf_linux_x86(self):
|
|
filename = 'test.qlog'
|
|
|
|
ql = Qiling(["../examples/rootfs/x86_linux/bin/x86_hello"], "../examples/rootfs/x86_linux", verbose=QL_VERBOSE.DEBUG, log_devices=[filename])
|
|
ql.run()
|
|
ql._log_file_fd.handlers[0].close() # prevent FD leak that causes downstream issues
|
|
os.remove(filename)
|
|
del ql
|
|
|
|
def test_elf_linux_x86_static(self):
|
|
ql = Qiling(["../examples/rootfs/x86_linux/bin/x86_hello_static"], "../examples/rootfs/x86_linux", verbose=QL_VERBOSE.DEBUG)
|
|
ql.run()
|
|
del ql
|
|
|
|
def posix_syscall_test(self, argv: str, rootfs: str, syscalls: Sequence[str]):
|
|
"""A generic method to test out POSIX system calls hooking.
|
|
"""
|
|
|
|
checklist = []
|
|
|
|
def test_syscall_read(ql: Qiling, fd: int, buf: int, count: int):
|
|
retval = syscall.ql_syscall_read(ql, fd, buf, count)
|
|
|
|
hpath = ql.os.fd[fd].name
|
|
|
|
if os.path.basename(hpath) == "test_syscall_read.txt":
|
|
mcontent = ql.mem.read(buf, count)
|
|
|
|
with open(hpath, 'rb') as infile:
|
|
fcontent = infile.read()
|
|
|
|
if ql.host.os is not QL_OS.WINDOWS:
|
|
os.remove(hpath)
|
|
|
|
self.assertEqual(mcontent, fcontent)
|
|
checklist.append('read')
|
|
|
|
return retval
|
|
|
|
def test_syscall_write(ql: Qiling, fd: int, buf: int, count: int):
|
|
retval = syscall.ql_syscall_write(ql, fd, buf, count)
|
|
|
|
hpath = ql.os.fd[fd].name
|
|
|
|
if os.path.basename(hpath) == "test_syscall_write.txt":
|
|
mcontent = ql.mem.read(buf, count)
|
|
|
|
with open(hpath, 'rb') as infile:
|
|
fcontent = infile.read()
|
|
|
|
if ql.host.os is not QL_OS.WINDOWS:
|
|
os.remove(hpath)
|
|
|
|
self.assertEqual(mcontent, fcontent)
|
|
checklist.append('write')
|
|
|
|
return retval
|
|
|
|
def test_syscall_open(ql: Qiling, path: int, flags: int, mode: int):
|
|
retval = syscall.ql_syscall_open(ql, path, flags, mode)
|
|
|
|
vpath = ql.os.utils.read_cstring(path)
|
|
|
|
if vpath == "test_syscall_open.txt":
|
|
hpath = ql.os.path.virtual_to_host_path(vpath)
|
|
|
|
self.assertTrue(os.path.isfile(hpath))
|
|
checklist.append('open')
|
|
|
|
if ql.host.os is not QL_OS.WINDOWS:
|
|
os.remove(hpath)
|
|
|
|
return retval
|
|
|
|
def test_syscall_openat(ql: Qiling, fd: int, path: int, flags: int, mode: int):
|
|
retval = syscall.ql_syscall_openat(ql, fd, path, flags, mode)
|
|
|
|
vpath = ql.os.utils.read_cstring(path)
|
|
|
|
if vpath == "test_syscall_open.txt":
|
|
hpath = ql.os.path.virtual_to_host_path(vpath)
|
|
|
|
self.assertTrue(os.path.isfile(hpath))
|
|
checklist.append('openat')
|
|
|
|
if ql.host.os is not QL_OS.WINDOWS:
|
|
os.remove(hpath)
|
|
|
|
return retval
|
|
|
|
def test_syscall_unlink(ql: Qiling, path: int):
|
|
retval = syscall.ql_syscall_unlink(ql, path)
|
|
|
|
vpath = ql.os.utils.read_cstring(path)
|
|
|
|
if vpath == "test_syscall_unlink.txt":
|
|
hpath = ql.os.path.virtual_to_host_path(vpath)
|
|
|
|
self.assertFalse(os.path.isfile(hpath))
|
|
checklist.append('unlink')
|
|
|
|
return retval
|
|
|
|
def test_syscall_unlinkat(ql: Qiling, fd: int, path: int, flags: int):
|
|
retval = syscall.ql_syscall_unlinkat(ql, fd, path, flags)
|
|
|
|
vpath = ql.os.utils.read_cstring(path)
|
|
|
|
if vpath == "test_syscall_unlink.txt":
|
|
hpath = ql.os.path.virtual_to_host_path(vpath)
|
|
|
|
self.assertFalse(os.path.isfile(hpath))
|
|
checklist.append('unlinkat')
|
|
|
|
return retval
|
|
|
|
def test_syscall_truncate(ql: Qiling, path: int, length: int):
|
|
retval = syscall.ql_syscall_truncate(ql, path, length)
|
|
|
|
vpath = ql.os.utils.read_cstring(path)
|
|
|
|
if vpath == "test_syscall_truncate.txt":
|
|
hpath = ql.os.path.virtual_to_host_path(vpath)
|
|
|
|
self.assertEqual(length, os.stat(hpath).st_size)
|
|
checklist.append('truncate')
|
|
|
|
if ql.host.os is not QL_OS.WINDOWS:
|
|
os.remove(hpath)
|
|
|
|
return retval
|
|
|
|
def test_syscall_ftruncate(ql: Qiling, fd: int, length: int):
|
|
retval = syscall.ql_syscall_ftruncate(ql, fd, length)
|
|
|
|
hpath = ql.os.fd[fd].name
|
|
|
|
if os.path.basename(hpath) == "test_syscall_ftruncate.txt":
|
|
self.assertEqual(length, os.stat(hpath).st_size)
|
|
checklist.append('ftruncate')
|
|
|
|
if ql.host.os is not QL_OS.WINDOWS:
|
|
os.remove(hpath)
|
|
|
|
return retval
|
|
|
|
hooks = {
|
|
'read' : test_syscall_read,
|
|
'write' : test_syscall_write,
|
|
'open' : test_syscall_open,
|
|
'openat' : test_syscall_openat,
|
|
'unlink' : test_syscall_unlink,
|
|
'unlinkat' : test_syscall_unlinkat,
|
|
'truncate' : test_syscall_truncate,
|
|
'ftruncate' : test_syscall_ftruncate,
|
|
}
|
|
|
|
ql = Qiling([f'{rootfs}{argv}'], rootfs, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
# hook reuested system calls
|
|
for name in syscalls:
|
|
ql.os.set_syscall(name, hooks[name])
|
|
|
|
ql.run()
|
|
|
|
# make sure we visited them all
|
|
self.assertSequenceEqual(syscalls, checklist)
|
|
|
|
def test_elf_linux_x86_posix_syscall(self):
|
|
syscalls = ['openat', 'write', 'read', 'truncate', 'ftruncate', 'unlink']
|
|
|
|
self.posix_syscall_test(r'/bin/x86_posix_syscall', r'../examples/rootfs/x86_linux', syscalls)
|
|
|
|
def test_elf_linux_arm(self):
|
|
def my_puts(ql: Qiling):
|
|
params = ql.os.resolve_fcall_params(ELFTest.PARAMS_PUTS)
|
|
print(f'puts("{params["s"]}")')
|
|
|
|
all_mem = ql.mem.save()
|
|
ql.mem.restore(all_mem)
|
|
|
|
ql = Qiling(["../examples/rootfs/arm_linux/bin/arm_hello"], "../examples/rootfs/arm_linux", verbose=QL_VERBOSE.DEBUG)
|
|
ql.os.set_api('puts', my_puts)
|
|
ql.run()
|
|
del ql
|
|
|
|
def test_setsockopt_mips_so_rcvbuf(self):
|
|
# MIPS uses its own SO_* numbering (SO_RCVBUF = 0x1002, not the generic
|
|
# 0x08). The MIPS socket-option table had the wrong values, so e.g.
|
|
# busybox ping's setsockopt(SOL_SOCKET, SO_RCVBUF, ...) raised
|
|
# NotImplementedError ("Could not convert emulated socket option 4098").
|
|
from qiling.const import QL_ENDIAN
|
|
from qiling.os.posix.syscall.socket import ql_syscall_socket, ql_syscall_setsockopt
|
|
|
|
ql = Qiling(code=b"\x00\x00\x00\x00", archtype=QL_ARCH.MIPS, ostype=QL_OS.LINUX,
|
|
endian=QL_ENDIAN.EB, rootfs="../examples/rootfs/mips32_linux",
|
|
verbose=QL_VERBOSE.OFF)
|
|
|
|
AF_INET, SOCK_DGRAM = 2, 2
|
|
SOL_SOCKET, SO_RCVBUF = 0xffff, 0x1002 # MIPS values
|
|
|
|
fd = ql_syscall_socket(ql, AF_INET, SOCK_DGRAM, 0)
|
|
self.assertGreaterEqual(fd, 0)
|
|
|
|
base = 0x100000
|
|
ql.mem.map(base, 0x1000)
|
|
ql.mem.write_ptr(base, 16384, 4) # requested SO_RCVBUF size
|
|
|
|
# must map SO_RCVBUF (0x1002) to the host option and succeed, not raise
|
|
self.assertEqual(ql_syscall_setsockopt(ql, fd, SOL_SOCKET, SO_RCVBUF, base, 4), 0)
|
|
|
|
del ql
|
|
|
|
def test_elf_linux_arm_static(self):
|
|
ql = Qiling(["../examples/rootfs/arm_linux/bin/arm_hello_static"], "../examples/rootfs/arm_linux", verbose=QL_VERBOSE.DEFAULT)
|
|
all_mem = ql.mem.save()
|
|
ql.mem.restore(all_mem)
|
|
ql.run()
|
|
del ql
|
|
|
|
@unittest.skip('broken: ARM executable should be generated properly')
|
|
def test_elf_linux_arm_posix_syscall(self):
|
|
# TODO: check the list, it might be inacurate
|
|
syscalls = ['openat', 'write', 'read', 'truncate', 'ftruncate', 'unlink']
|
|
|
|
self.posix_syscall_test(r'/bin/arm_posix_syscall', r'../examples/rootfs/arm_linux', syscalls)
|
|
|
|
def test_elf_linux_arm64(self):
|
|
ql = Qiling(["../examples/rootfs/arm64_linux/bin/arm64_hello"], "../examples/rootfs/arm64_linux", verbose=QL_VERBOSE.DEBUG)
|
|
ql.run()
|
|
del ql
|
|
|
|
def test_elf_linux_arm64_static(self):
|
|
ql = Qiling(["../examples/rootfs/arm64_linux/bin/arm64_hello_static"], "../examples/rootfs/arm64_linux", verbose=QL_VERBOSE.DEFAULT)
|
|
ql.run()
|
|
del ql
|
|
|
|
def test_elf_linux_mips32eb_static(self):
|
|
ql = Qiling(["../examples/rootfs/mips32_linux/bin/mips32_hello_static"], "../examples/rootfs/mips32_linux")
|
|
ql.run()
|
|
del ql
|
|
|
|
@staticmethod
|
|
def random_generator(length: int):
|
|
chars = string.ascii_uppercase + string.digits
|
|
|
|
return ''.join(random.choices(chars, k=length))
|
|
|
|
def test_elf_linux_mips32eb(self):
|
|
ql = Qiling(["../examples/rootfs/mips32_linux/bin/mips32_hello", self.random_generator(64)], "../examples/rootfs/mips32_linux")
|
|
ql.run()
|
|
|
|
del ql
|
|
|
|
def test_mips32eb_fake_urandom(self):
|
|
class Fake_urandom(QlFsMappedObject):
|
|
def read(self, size: int):
|
|
return b'\x01' * size
|
|
|
|
def fstat(self):
|
|
return -1
|
|
|
|
def close(self):
|
|
return 0
|
|
|
|
ql = Qiling(["../examples/rootfs/mips32_linux/bin/mips32_fetch_urandom"], "../examples/rootfs/mips32_linux")
|
|
ql.add_fs_mapper("/dev/urandom", Fake_urandom())
|
|
|
|
ql.run()
|
|
del ql
|
|
|
|
def test_elf_onEnter_mips32el(self):
|
|
def my_puts_onenter(ql: Qiling):
|
|
params = ql.os.resolve_fcall_params(ELFTest.PARAMS_PUTS)
|
|
print(f'puts("{params["s"]}")')
|
|
|
|
params = ql.os.fcall.readParams(ELFTest.PARAMS_PUTS.values())
|
|
self.my_puts_onenter_addr = params[0]
|
|
|
|
return 2
|
|
|
|
ql = Qiling(["../examples/rootfs/mips32el_linux/bin/mips32el_double_hello"], "../examples/rootfs/mips32el_linux")
|
|
ql.os.set_api('puts', my_puts_onenter, QL_INTERCEPT.ENTER)
|
|
ql.run()
|
|
|
|
self.assertEqual(4196680, self.my_puts_onenter_addr)
|
|
|
|
del ql
|
|
|
|
def test_elf_linux_arm64_posix_syscall(self):
|
|
syscalls = ['openat', 'write', 'read', 'truncate', 'ftruncate', 'unlinkat']
|
|
|
|
self.posix_syscall_test(r'/bin/arm64_posix_syscall', r'../examples/rootfs/arm64_linux', syscalls)
|
|
|
|
def test_elf_linux_mips32el(self):
|
|
ql = Qiling(["../examples/rootfs/mips32el_linux/bin/mips32el_hello", self.random_generator(64)], "../examples/rootfs/mips32el_linux")
|
|
ql.run()
|
|
del ql
|
|
|
|
def test_elf_linux_mips32el_static(self):
|
|
ql = Qiling(["../examples/rootfs/mips32el_linux/bin/mips32el_hello_static", self.random_generator(64)], "../examples/rootfs/mips32el_linux")
|
|
ql.run()
|
|
del ql
|
|
|
|
def test_elf_linux_mips32el_posix_syscall(self):
|
|
syscalls = ['open', 'write', 'read', 'truncate', 'ftruncate', 'unlink']
|
|
|
|
self.posix_syscall_test(r'/bin/mips32el_posix_syscall', r'../examples/rootfs/mips32el_linux', syscalls)
|
|
|
|
def test_elf_linux_powerpc(self):
|
|
ql = Qiling(["../examples/rootfs/powerpc_linux/bin/powerpc_hello"], "../examples/rootfs/powerpc_linux", verbose=QL_VERBOSE.DEBUG)
|
|
ql.run()
|
|
del ql
|
|
|
|
def test_elf_linux_arm_custom_syscall(self):
|
|
checklist = {}
|
|
|
|
def my_syscall_write(ql: Qiling, fd: int, buf: int, count: int):
|
|
try:
|
|
data = ql.mem.read(buf, count)
|
|
ql.os.fd[fd].write(data)
|
|
except:
|
|
regreturn = -1
|
|
else:
|
|
regreturn = count
|
|
|
|
checklist['set_syscall'] = ql.arch.regs.r0
|
|
|
|
return regreturn
|
|
|
|
ql = Qiling(["../examples/rootfs/arm_linux/bin/arm_hello"], "../examples/rootfs/arm_linux")
|
|
ql.os.set_syscall('write', my_syscall_write)
|
|
ql.run()
|
|
|
|
self.assertEqual(1, checklist['set_syscall'])
|
|
|
|
del ql
|
|
|
|
def test_elf_linux_x86_crackme(self):
|
|
def instruction_count(ql: Qiling, address: int, size: int, user_data: Any):
|
|
user_data[0] += 1
|
|
|
|
def my__llseek(ql, *args, **kw):
|
|
pass
|
|
|
|
def run_one_round(payload):
|
|
ql = Qiling(["../examples/rootfs/x86_linux/bin/crackme_linux"], "../examples/rootfs/x86_linux", console=False)
|
|
|
|
ins_count = [0]
|
|
ql.hook_code(instruction_count, ins_count)
|
|
ql.os.set_syscall("_llseek", my__llseek)
|
|
|
|
ql.os.stdin = pipe.SimpleInStream(0)
|
|
ql.os.stdin.write(payload)
|
|
|
|
ql.run()
|
|
del ql
|
|
|
|
return ins_count[0]
|
|
|
|
def solve():
|
|
idx_list = [1, 4, 2, 0, 3]
|
|
|
|
flag = b'\x00\x00\x00\x00\x00\n'
|
|
|
|
old_count = run_one_round(flag)
|
|
for idx in idx_list:
|
|
for i in b'L1NUX\\n':
|
|
flag = flag[:idx] + chr(i).encode() + flag[idx + 1:]
|
|
tmp = run_one_round(flag)
|
|
|
|
if tmp > old_count:
|
|
old_count = tmp
|
|
break
|
|
|
|
print(flag)
|
|
|
|
print("Linux Simple Crackme Brute Force, This Will Take Some Time ...")
|
|
solve()
|
|
|
|
def test_x86_fake_urandom_multiple_times(self):
|
|
next_id = 0
|
|
|
|
def get_next_id() -> int:
|
|
nonlocal next_id
|
|
|
|
curr_id = next_id
|
|
next_id += 1
|
|
|
|
return curr_id
|
|
|
|
ids = []
|
|
|
|
class Fake_urandom(QlFsMappedObject):
|
|
def __init__(self):
|
|
ids.append(get_next_id())
|
|
|
|
def read(self, size: int):
|
|
return b'\x01' * size
|
|
|
|
def fstat(self):
|
|
return -1
|
|
|
|
def close(self):
|
|
return 0
|
|
|
|
ql = Qiling(["../examples/rootfs/x86_linux/bin/x86_fetch_urandom_multiple_times"], "../examples/rootfs/x86_linux")
|
|
ql.add_fs_mapper("/dev/urandom", Fake_urandom())
|
|
|
|
ql.run()
|
|
|
|
self.assertListEqual([0], ids)
|
|
|
|
del ql
|
|
|
|
def test_x86_fake_urandom(self):
|
|
class Fake_urandom(QlFsMappedObject):
|
|
|
|
def read(self, size: int):
|
|
return b"\x01" * size
|
|
|
|
def fstat(self):
|
|
return -1
|
|
|
|
def close(self):
|
|
return 0
|
|
|
|
ql = Qiling(["../examples/rootfs/x86_linux/bin/x86_fetch_urandom"], "../examples/rootfs/x86_linux", verbose=QL_VERBOSE.DEBUG)
|
|
ql.add_fs_mapper("/dev/urandom", Fake_urandom())
|
|
|
|
ql.run()
|
|
del ql
|
|
|
|
def test_x8664_map_urandom(self):
|
|
ql = Qiling(["../examples/rootfs/x8664_linux/bin/x8664_fetch_urandom"], "../examples/rootfs/x8664_linux", verbose=QL_VERBOSE.DEBUG)
|
|
ql.add_fs_mapper("/dev/urandom", "/dev/urandom")
|
|
|
|
ql.run()
|
|
|
|
del ql
|
|
|
|
def test_x8664_symlink(self):
|
|
ql = Qiling(["../examples/rootfs/x8664_linux_symlink/bin/x8664_hello"], "../examples/rootfs/x8664_linux_symlink", verbose=QL_VERBOSE.DEBUG)
|
|
ql.run()
|
|
del ql
|
|
|
|
def test_x8664_absolute_path(self):
|
|
ql = Qiling(["../examples/rootfs/x8664_linux/bin/absolutepath"], "../examples/rootfs/x8664_linux", verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stdout = pipe.SimpleOutStream(1)
|
|
ql.run()
|
|
|
|
self.assertEqual(ql.os.stdout.read(), b'test_complete\n\ntest_complete\n\n')
|
|
|
|
del ql
|
|
|
|
def test_x8664_getcwd(self):
|
|
ql = Qiling(["../examples/rootfs/x8664_linux/bin/testcwd"], "../examples/rootfs/x8664_linux", verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stdout = pipe.SimpleOutStream(1)
|
|
ql.run()
|
|
|
|
self.assertEqual(ql.os.stdout.read(), b'/\n/lib\n/bin\n/\n')
|
|
|
|
del ql
|
|
|
|
def test_elf_linux_x86_return_from_main_stackpointer(self):
|
|
ql = Qiling(["../examples/rootfs/x86_linux/bin/x86_return_main"], "../examples/rootfs/x86_linux", stop=QL_STOP.STACK_POINTER)
|
|
ql.run()
|
|
del ql
|
|
|
|
def test_elf_linux_x86_return_from_main_exit_trap(self):
|
|
ql = Qiling(["../examples/rootfs/x86_linux/bin/x86_return_main"], "../examples/rootfs/x86_linux", stop=QL_STOP.EXIT_TRAP)
|
|
ql.run()
|
|
del ql
|
|
|
|
def test_elf_linux_x8664_return_from_main_stackpointer(self):
|
|
ql = Qiling(["../examples/rootfs/x8664_linux/bin/x8664_return_main"], "../examples/rootfs/x8664_linux", stop=QL_STOP.STACK_POINTER)
|
|
ql.run()
|
|
del ql
|
|
|
|
def test_elf_linux_x8664_return_from_main_exit_trap(self):
|
|
ql = Qiling(["../examples/rootfs/x8664_linux/bin/x8664_return_main"], "../examples/rootfs/x8664_linux", stop=QL_STOP.EXIT_TRAP)
|
|
ql.run()
|
|
del ql
|
|
|
|
def test_arm_stat64(self):
|
|
ql = Qiling(["../examples/rootfs/arm_linux/bin/arm_stat64", "/bin/arm_stat64"], "../examples/rootfs/arm_linux", verbose=QL_VERBOSE.DEBUG)
|
|
ql.run()
|
|
del ql
|
|
|
|
def test_elf_linux_x8664_getdents(self):
|
|
ql = Qiling(["../examples/rootfs/x8664_linux/bin/x8664_getdents"], "../examples/rootfs/x8664_linux", verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stdout = io.BytesIO()
|
|
ql.run()
|
|
|
|
ql.os.stdout.seek(0)
|
|
self.assertIn("bin\n", ql.os.stdout.read().decode("utf-8"))
|
|
|
|
del ql
|
|
|
|
def test_elf_linux_armeb(self):
|
|
ql = Qiling(["../examples/rootfs/armeb_linux/bin/armeb_hello"], "../examples/rootfs/armeb_linux", verbose=QL_VERBOSE.DEBUG)
|
|
ql.run()
|
|
del ql
|
|
|
|
def test_elf_linux_armeb_static(self):
|
|
ql = Qiling(["../examples/rootfs/armeb_linux/bin/armeb_hello_static"], "../examples/rootfs/armeb_linux", verbose=QL_VERBOSE.DEFAULT)
|
|
ql.run()
|
|
del ql
|
|
|
|
# TODO: Disable for now
|
|
# def test_armoabi_eb_linux_syscall_elf_static(self):
|
|
# # src: https://github.com/qilingframework/qiling/blob/1f1e9bc756e59a0bfc112d32735f8882b1afc165/examples/src/linux/posix_syscall.c
|
|
# ql = Qiling(["../examples/rootfs/armeb_linux/bin/posix_syscall_msb.armoabi"], "../examples/rootfs/armeb_linux", verbose=QL_VERBOSE.DEBUG)
|
|
# ql.run()
|
|
|
|
def test_armoabi_le_linux_syscall_elf_static(self):
|
|
# src: https://github.com/qilingframework/qiling/blob/1f1e9bc756e59a0bfc112d32735f8882b1afc165/examples/src/linux/posix_syscall.c
|
|
ql = Qiling(["../examples/rootfs/arm_linux/bin/posix_syscall_lsb.armoabi"], "../examples/rootfs/arm_linux", verbose=QL_VERBOSE.DEBUG)
|
|
ql.run()
|
|
del ql
|
|
|
|
def test_elf_linux_x86_getdents64(self):
|
|
ql = Qiling(["../examples/rootfs/x86_linux/bin/x86_getdents64"], "../examples/rootfs/x86_linux", verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stdout = pipe.SimpleOutStream(1)
|
|
ql.run()
|
|
|
|
self.assertIn("bin\n", ql.os.stdout.read().decode("utf-8"))
|
|
|
|
del ql
|
|
|
|
# Regression for getdents64 record alignment. Each linux_dirent64 record
|
|
# must be padded so the next record's leading u64 d_ino stays 8-byte
|
|
# aligned; otherwise a strict-alignment guest (e.g. MIPS) faults with an
|
|
# unaligned load while walking the buffer. x86 tolerates the unaligned read,
|
|
# which is why it was never caught.
|
|
def test_linux_getdents64_alignment(self):
|
|
from qiling.const import QL_ENDIAN
|
|
from qiling.os.posix.syscall.fcntl import ql_syscall_open
|
|
from qiling.os.posix.syscall.unistd import ql_syscall_getdents64
|
|
|
|
ql = Qiling(code=b"\x00\x00\x00\x00", archtype=QL_ARCH.MIPS, ostype=QL_OS.LINUX,
|
|
endian=QL_ENDIAN.EB, rootfs="../examples/rootfs/mips32_linux",
|
|
verbose=QL_VERBOSE.OFF)
|
|
|
|
base = 0x100000
|
|
ql.mem.map(base, 0x4000)
|
|
path_ptr, buf_ptr = base, base + 0x100
|
|
ql.mem.write(path_ptr, b"/\x00")
|
|
|
|
fd = ql_syscall_open(ql, path_ptr, 0, 0) # O_RDONLY on a directory
|
|
self.assertGreaterEqual(fd, 0)
|
|
|
|
nbytes = ql_syscall_getdents64(ql, fd, buf_ptr, 0x2000)
|
|
self.assertGreater(nbytes, 0)
|
|
|
|
# walk the linux_dirent64 records; every record (hence its leading u64
|
|
# d_ino) must start at an 8-byte boundary. d_reclen is a u16 at offset 16.
|
|
buf = bytes(ql.mem.read(buf_ptr, nbytes))
|
|
off = 0
|
|
while off < nbytes:
|
|
self.assertEqual(off % 8, 0, f"dirent at offset {off} is not 8-byte aligned")
|
|
d_reclen = int.from_bytes(buf[off + 16:off + 18], "big")
|
|
self.assertNotEqual(d_reclen, 0)
|
|
off += d_reclen
|
|
|
|
del ql
|
|
|
|
def test_linux_getdents_alignment(self):
|
|
# the legacy getdents record was not rounded up to the alignment of its
|
|
# word-sized d_ino either, so a strict-alignment guest faulted walking
|
|
# the buffer. unlike getdents64, legacy linux_dirent keeps d_type in the
|
|
# record's last byte (offset d_reclen-1), so the padding must sit between
|
|
# d_name and d_type -- verify the records are aligned AND d_type survives.
|
|
from qiling.const import QL_ENDIAN
|
|
from qiling.os.posix.syscall.fcntl import ql_syscall_open
|
|
from qiling.os.posix.syscall.unistd import ql_syscall_getdents
|
|
|
|
ql = Qiling(code=b"\x00\x00\x00\x00", archtype=QL_ARCH.MIPS, ostype=QL_OS.LINUX,
|
|
endian=QL_ENDIAN.EB, rootfs="../examples/rootfs/mips32_linux",
|
|
verbose=QL_VERBOSE.OFF)
|
|
|
|
base = 0x100000
|
|
ql.mem.map(base, 0x4000)
|
|
path_ptr, buf_ptr = base, base + 0x100
|
|
ql.mem.write(path_ptr, b"/\x00")
|
|
|
|
fd = ql_syscall_open(ql, path_ptr, 0, 0) # O_RDONLY on a directory
|
|
self.assertGreaterEqual(fd, 0)
|
|
|
|
nbytes = ql_syscall_getdents(ql, fd, buf_ptr, 0x2000)
|
|
self.assertGreater(nbytes, 0)
|
|
|
|
# walk the linux_dirent records; on 32-bit MIPS d_ino is 4 bytes, so each
|
|
# record must start at a 4-byte boundary. d_reclen is a u16 at offset 8
|
|
# (after the 4-byte d_ino + 4-byte d_off); d_name starts at offset 10.
|
|
buf = bytes(ql.mem.read(buf_ptr, nbytes))
|
|
off = 0
|
|
while off < nbytes:
|
|
self.assertEqual(off % 4, 0, f"dirent at offset {off} is not 4-byte aligned")
|
|
d_reclen = int.from_bytes(buf[off + 8:off + 10], "big")
|
|
self.assertNotEqual(d_reclen, 0)
|
|
|
|
name = buf[off + 10:buf.index(b"\x00", off + 10)].decode()
|
|
# legacy getdents stores d_type in the record's last byte; '.' and
|
|
# '..' must still report DT_DIR (4) after the alignment padding
|
|
if name in (".", ".."):
|
|
self.assertEqual(buf[off + d_reclen - 1], 4, f"{name}: d_type not DT_DIR")
|
|
|
|
off += d_reclen
|
|
|
|
self.assertEqual(off, nbytes) # records tile the buffer exactly
|
|
|
|
del ql
|
|
|
|
# Regression for statx() byte order on big-endian guests: the statx struct
|
|
# must be serialized in the guest's endianness. Otherwise stx_mode is byte-
|
|
# swapped and a directory loses its S_IFDIR bit (so e.g. `ls` treats a
|
|
# directory as a regular file). Exercised on a big-endian MIPS context.
|
|
def test_linux_statx_bigendian(self):
|
|
import stat as _stat
|
|
from qiling.const import QL_ENDIAN
|
|
from qiling.os.posix.syscall.stat import ql_syscall_statx, Statx32
|
|
|
|
ql = Qiling(code=b"\x00\x00\x00\x00", archtype=QL_ARCH.MIPS, ostype=QL_OS.LINUX,
|
|
endian=QL_ENDIAN.EB, rootfs="../examples/rootfs/mips32_linux",
|
|
verbose=QL_VERBOSE.OFF)
|
|
|
|
base = 0x100000
|
|
ql.mem.map(base, 0x1000)
|
|
path_ptr, buf_ptr = base, base + 0x200
|
|
ql.mem.write(path_ptr, b"/\x00")
|
|
|
|
AT_FDCWD = -100 & 0xffffffff
|
|
STATX_BASIC_STATS = 0x07ff
|
|
ret = ql_syscall_statx(ql, AT_FDCWD, path_ptr, 0, STATX_BASIC_STATS, buf_ptr)
|
|
self.assertEqual(ret, 0)
|
|
|
|
# the guest is big-endian, so it reads stx_mode in big-endian byte order;
|
|
# it must come back as a directory (S_IFDIR). Before the fix the struct
|
|
# was emitted little-endian and the type bits were lost.
|
|
mode = int.from_bytes(ql.mem.read(buf_ptr + Statx32.stx_mode.offset, 2), 'big')
|
|
self.assertTrue(_stat.S_ISDIR(mode))
|
|
|
|
del ql
|
|
|
|
def test_memory_search(self):
|
|
ql = Qiling(code=b"\xCC", archtype=QL_ARCH.X8664, ostype=QL_OS.LINUX, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.mem.map(0x1000, 0x1000)
|
|
ql.mem.map(0x2000, 0x1000)
|
|
ql.mem.map(0x3000, 0x1000)
|
|
|
|
ql.mem.write(0x1000, b"\x47\x06\x0d\x1e\x0d\x1a\x53\x0f\x07\x06\x06\x09\x53\x0f\x01\x1e\x0d\x53\x11\x07\x1d\x53\x1d\x18\x4f\x53\x06\x0d\x1e\x0d\x1a\x53\x0f\x07\x06\x06\x09\x53\x04\x0d\x1c\x53\x11\x07\x1d\x53\x0c\x07\x1f\x06\x45")
|
|
ql.mem.write(0x2000, b"\x47\x06\x0d\x1e\x0d\x1a\x53\x0f\x07\x06\x06\x09\x53\x1a\x1d\x06\x53\x09\x1a\x07\x1d\x06\x0c\x53\x09\x06\x0c\x53\x0c\x0d\x1b\x0d\x1a\x1c\x53\x11\x07\x1d\x4f\x53\x06\x0d\x1e\x0d\x1a\x53\x0f\x07\x06\x06\x09\x53\x05\x09\x03\x0d\x53\x11\x07\x1d\x53\x0b\x1a\x11\x45")
|
|
ql.mem.write(0x3000, b"\x47\x06\x0d\x1e\x0d\x1a\x53\x0f\x07\x06\x06\x09\x53\x1b\x09\x11\x53\x0f\x07\x07\x0c\x0a\x11\x0d\x4f\x53\x06\x0d\x1e\x0d\x1a\x53\x0f\x07\x06\x06\x09\x53\x1c\x0d\x04\x04\x53\x09\x53\x04\x01\x0d\x53\x09\x06\x0c\x53\x00\x1d\x1a\x1c\x53\x11\x07\x1d\x45")
|
|
ql.mem.write(0x1FFB, b"\x1f\x00\x07\x53\x03\x06\x07\x1f\x1b")
|
|
|
|
# Needle not in haystack
|
|
self.assertEqual([], ql.mem.search(b"\x3a\x01\x0b\x03\x53\x29\x1b\x1c\x04\x0d\x11"))
|
|
|
|
# Needle appears several times in haystack
|
|
self.assertEqual([0x1000 + 24, 0x2000 + 38, 0x3000 + 24], ql.mem.search(b"\x4f\x53\x06\x0d\x1e\x0d\x1a"))
|
|
|
|
# Needle inside haystack
|
|
self.assertEqual([0x1000 + 13], ql.mem.search(b"\x0f\x01\x1e\x0d\x53\x11\x07\x1d\x53\x1d\x18", begin=0x1000 + 10, end=0x1000 + 30))
|
|
|
|
# Needle before haystack
|
|
self.assertEqual([], ql.mem.search(b"\x04\x0d\x1c\x53\x11\x07\x1d\x53\x0c\x07\x1f\x06", begin=0x1337))
|
|
|
|
# Needle after haystack
|
|
self.assertEqual([], ql.mem.search(b"\x1b\x09\x11\x53\x0f\x07\x07\x0c\x0a\x11\x0d", end=0x3000 + 13))
|
|
|
|
# Needle exactly inside haystack
|
|
self.assertEqual([0x2000 + 13], ql.mem.search(b"\x1a\x1d\x06\x53\x09\x1a\x07\x1d\x06\x0c", begin=0x2000 + 13, end=0x2000 + 23))
|
|
|
|
# Needle 'tears' two mapped regions
|
|
self.assertEqual([], ql.mem.search(b"\x1f\x00\x07\x53\x03\x06\x07\x1f\x1b", begin=0x1F00, end=0x200F))
|
|
|
|
# Needle is a regex
|
|
self.assertEqual([0x1000 + 11, 0x2000 + 11, 0x3000 + 43], ql.mem.search(re.compile(b"\x09\x53(\x0f|\x1a|\x04)[^\x0d]")))
|
|
|
|
del ql
|
|
|
|
def test_elf_linux_x8664_path_traversion(self):
|
|
ql = Qiling(["../examples/rootfs/x8664_linux/bin/path_traverse_static"], "../examples/rootfs/x8664_linux", verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stdout = pipe.SimpleOutStream(1)
|
|
ql.run()
|
|
|
|
self.assertNotIn("root\n", ql.os.stdout.read().decode("utf-8"))
|
|
|
|
del ql
|
|
|
|
@unittest.skip("stdin hijacking doesn't work as expected")
|
|
def test_elf_linux_x8664_epoll_simple(self):
|
|
# This tests a sample binary that (e)polls on stdin and echos back the output. Upon
|
|
# receiving 'stop', it will exit.
|
|
#
|
|
# epoll-0 tkaen from: https://github.com/maxasm/epoll-c/blob/main/main.c
|
|
|
|
rootfs = "../examples/rootfs/x8664_linux"
|
|
argv = r"../examples/rootfs/x8664_linux/bin/x8664_linux_epoll_0".split()
|
|
ql = Qiling(argv, rootfs, verbose=QL_VERBOSE.DEBUG)
|
|
ql.os.stdin = pipe.SimpleInStream(0)
|
|
ql.os.stdin.write(b'echo\n')
|
|
ql.os.stdin.write(b'stop\n') # signal to exit gracefully
|
|
ql.run()
|
|
|
|
self.assertIn(b'echo\n', ql.os.stdout.read())
|
|
|
|
del ql
|
|
|
|
def test_elf_linux_x8664_epoll_server(self):
|
|
# This tests a simple server that uses epoll to wait for data, then prints it out. It has
|
|
# been modified to exit after data has been received; instead of a typical server operation
|
|
# that reads requests indefinitely.
|
|
#
|
|
# It listens on port 8000, and a separate thread is spawned in order to test how the server
|
|
# handles a 'hello world' input. The server prints out whatever it receives, so the assert
|
|
# statement checks the input is present as expected.
|
|
#
|
|
# onestraw server taken from: https://github.com/onestraw/epoll-example
|
|
|
|
# Note: Without a hook for this syscall, this error fires:
|
|
# TypeError: stat: path should be string, bytes, os.PathLike or integer, not NoneType
|
|
def hook_newfstatat(ql: Qiling, dirfd: int, pathname: int, statbuf: int, flags: int):
|
|
return 0
|
|
|
|
def client():
|
|
# give time for the server to listen
|
|
time.sleep(3)
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
s.connect(("127.0.0.1", 8000))
|
|
s.send(b"hello world")
|
|
s.close()
|
|
|
|
rootfs = "../examples/rootfs/x8664_linux"
|
|
argv = r"../examples/rootfs/x8664_linux/bin/x8664_onestraw_server s".split() # s means 'server mode'
|
|
|
|
ql = Qiling(argv, rootfs, verbose=QL_VERBOSE.DEBUG)
|
|
ql.os.set_syscall("newfstatat", hook_newfstatat, QL_INTERCEPT.CALL)
|
|
ql.os.stdout = pipe.SimpleOutStream(1) # server prints data received to stdout
|
|
ql.filter = '^data:'
|
|
|
|
client_thread = threading.Thread(target=client, daemon=True)
|
|
client_thread.start()
|
|
ql.run()
|
|
|
|
self.assertIn(b'hello world', ql.os.stdout.read(200)) # 200 is arbitrary--"good enough" for this task
|
|
del ql
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|