mirror of
https://github.com/qilingframework/qiling.git
synced 2026-10-12 11:29:57 +00:00
retrocpugeekandClaude Opus 4.8
f5c0afab58
Modern glibc (recent toolchains) issues clone3 from pthread_create rather than clone, so thread creation fails on any guest built against it: Qiling does not return -ENOSYS for unimplemented syscalls (it logs a warning and leaves the return register untouched), so glibc's clone3->clone fallback never fires. Add ql_syscall_clone3, which unpacks struct clone_args and delegates to the existing clone() handler. Translations: child_stack = stack + stack_size (clone3 passes the stack base plus a size; legacy clone wants the highest address), exit_signal folded into the flags' CSIGNAL byte, and an x8664 pre-swap that cancels ql_syscall_clone's arch-specific newtls<->child_tidptr swap. Add a self-contained regression test (test_clone3_translates_to_clone) that drives ql_syscall_clone3 directly and asserts the translation for both the generic path and the x8664 swap. Runs on stock unicorn; no clone3 binary needed. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
690 lines
26 KiB
Python
690 lines
26 KiB
Python
#!/usr/bin/env python3
|
|
#
|
|
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
|
|
#
|
|
|
|
import http.client
|
|
import platform
|
|
import re
|
|
import sys
|
|
import os
|
|
import threading
|
|
import time
|
|
import unittest
|
|
import socket
|
|
from typing import List
|
|
|
|
sys.path.append("..")
|
|
from qiling import Qiling
|
|
from qiling.arch.models import X86_CPU_MODEL
|
|
from qiling.const import QL_VERBOSE, QL_INTERCEPT, QL_ARCH, QL_OS
|
|
from qiling.os.filestruct import ql_file
|
|
from qiling.os.stats import QlOsNullStats
|
|
import qiling.os.posix.syscall.sched as ql_sched
|
|
|
|
|
|
BASE_ROOTFS = r'../examples/rootfs'
|
|
X86_LINUX_ROOTFS = fr'{BASE_ROOTFS}/x86_linux'
|
|
X64_LINUX_ROOTFS = fr'{BASE_ROOTFS}/x8664_linux'
|
|
ARM_LINUX_ROOTFS = fr'{BASE_ROOTFS}/arm_linux'
|
|
ARMEB_LINUX_ROOTFS = fr'{BASE_ROOTFS}/armeb_linux'
|
|
ARM64_LINUX_ROOTFS = fr'{BASE_ROOTFS}/arm64_linux'
|
|
MIPSEB_LINUX_ROOTFS = fr'{BASE_ROOTFS}/mips32_linux'
|
|
MIPSEL_LINUX_ROOTFS = fr'{BASE_ROOTFS}/mips32el_linux'
|
|
|
|
|
|
class ELFTest(unittest.TestCase):
|
|
|
|
@unittest.skipIf(platform.system() == "Darwin" and platform.machine() == "arm64", 'darwin host')
|
|
def test_elf_linux_execve_x8664(self):
|
|
ql = Qiling([fr'{X64_LINUX_ROOTFS}/bin/posix_syscall_execve'], X64_LINUX_ROOTFS, verbose=QL_VERBOSE.DEBUG)
|
|
ql.run()
|
|
|
|
env = ql.loader.env
|
|
|
|
self.assertIn('QL_TEST', env)
|
|
self.assertEqual('TEST_QUERY', env['QL_TEST'])
|
|
self.assertEqual('child', ql.loader.argv[0])
|
|
|
|
def test_elf_linux_cloexec_x8664(self):
|
|
ql = Qiling([fr'{X64_LINUX_ROOTFS}/bin/x8664_cloexec_test'], X64_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
filename = 'stderr.txt'
|
|
err = ql_file.open(filename, os.O_RDWR | os.O_CREAT, 0o644)
|
|
|
|
ql.os.stats = QlOsNullStats()
|
|
ql.os.stderr = err
|
|
|
|
ql.run()
|
|
err.close()
|
|
|
|
with open(filename, 'r') as f:
|
|
contents = f.readlines()
|
|
|
|
# cleanup
|
|
os.remove(filename)
|
|
|
|
self.assertGreaterEqual(len(contents), 4)
|
|
self.assertIn('Bad file descriptor', contents[-2])
|
|
self.assertIn('Bad file descriptor', contents[-1])
|
|
|
|
def test_multithread_elf_linux_x86(self):
|
|
logged: List[str] = []
|
|
|
|
def check_write(ql: Qiling, fd: int, write_buf, count: int):
|
|
if fd == 1:
|
|
content = ql.mem.read(write_buf, count)
|
|
|
|
logged.extend(content.decode().splitlines())
|
|
|
|
ql = Qiling([fr'{X86_LINUX_ROOTFS}/bin/x86_multithreading'], X86_LINUX_ROOTFS, cputype=X86_CPU_MODEL.INTEL_HASWELL, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stats = QlOsNullStats()
|
|
ql.os.set_syscall("write", check_write, QL_INTERCEPT.ENTER)
|
|
ql.run()
|
|
|
|
self.assertGreaterEqual(len(logged), 2)
|
|
self.assertTrue(logged[-2].startswith('thread 1 ret val is'))
|
|
self.assertTrue(logged[-1].startswith('thread 2 ret val is'))
|
|
|
|
def test_multithread_elf_linux_arm64(self):
|
|
logged: List[str] = []
|
|
|
|
def check_write(ql: Qiling, fd: int, write_buf, count: int):
|
|
if fd == 1:
|
|
content = ql.mem.read(write_buf, count)
|
|
|
|
logged.extend(content.decode().splitlines())
|
|
|
|
ql = Qiling([fr'{ARM64_LINUX_ROOTFS}/bin/arm64_multithreading'], ARM64_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stats = QlOsNullStats()
|
|
ql.os.set_syscall("write", check_write, QL_INTERCEPT.ENTER)
|
|
ql.run()
|
|
|
|
self.assertGreaterEqual(len(logged), 2)
|
|
self.assertTrue(logged[-2].startswith('thread 1 ret val is'))
|
|
self.assertTrue(logged[-1].startswith('thread 2 ret val is'))
|
|
|
|
def test_multithread_elf_linux_x8664(self):
|
|
logged: List[str] = []
|
|
|
|
def check_write(ql: Qiling, fd: int, write_buf, count: int):
|
|
if fd == 1:
|
|
content = ql.mem.read(write_buf, count)
|
|
|
|
logged.extend(content.decode().splitlines())
|
|
|
|
ql = Qiling([fr'{X64_LINUX_ROOTFS}/bin/x8664_multithreading'], X64_LINUX_ROOTFS, cputype=X86_CPU_MODEL.INTEL_HASWELL, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stats = QlOsNullStats()
|
|
ql.os.set_syscall("write", check_write, QL_INTERCEPT.ENTER)
|
|
ql.run()
|
|
|
|
self.assertGreaterEqual(len(logged), 2)
|
|
self.assertTrue(logged[-2].startswith('thread 1 ret val is'))
|
|
self.assertTrue(logged[-1].startswith('thread 2 ret val is'))
|
|
|
|
def test_multithread_elf_linux_mips32eb(self):
|
|
logged: List[str] = []
|
|
|
|
def check_write(ql: Qiling, fd: int, write_buf, count: int):
|
|
if fd == 1:
|
|
content = ql.mem.read(write_buf, count)
|
|
|
|
logged.extend(content.decode().splitlines())
|
|
|
|
ql = Qiling([fr'{MIPSEB_LINUX_ROOTFS}/bin/mips32_multithreading'], MIPSEB_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stats = QlOsNullStats()
|
|
ql.os.set_syscall("write", check_write, QL_INTERCEPT.ENTER)
|
|
ql.run()
|
|
|
|
self.assertGreaterEqual(len(logged), 2)
|
|
self.assertTrue(logged[-2].startswith('thread 1 ret val is'))
|
|
self.assertTrue(logged[-1].startswith('thread 2 ret val is'))
|
|
|
|
def test_multithread_elf_linux_mips32el(self):
|
|
logged: List[str] = []
|
|
|
|
def check_write(ql: Qiling, fd: int, write_buf, count: int):
|
|
if fd == 1:
|
|
content = ql.mem.read(write_buf, count)
|
|
|
|
logged.extend(content.decode().splitlines())
|
|
|
|
ql = Qiling([fr'{MIPSEL_LINUX_ROOTFS}/bin/mips32el_multithreading'], MIPSEL_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stats = QlOsNullStats()
|
|
ql.os.set_syscall("write", check_write, QL_INTERCEPT.ENTER)
|
|
ql.run()
|
|
|
|
self.assertGreaterEqual(len(logged), 2)
|
|
self.assertTrue(logged[-2].startswith('thread 1 ret val is'))
|
|
self.assertTrue(logged[-1].startswith('thread 2 ret val is'))
|
|
|
|
def test_multithread_elf_linux_arm(self):
|
|
logged: List[str] = []
|
|
|
|
def check_write(ql: Qiling, fd: int, write_buf, count: int):
|
|
if fd == 1:
|
|
content = ql.mem.read(write_buf, count)
|
|
|
|
logged.extend(content.decode().splitlines())
|
|
|
|
ql = Qiling([fr'{ARM_LINUX_ROOTFS}/bin/arm_multithreading'], ARM_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stats = QlOsNullStats()
|
|
ql.os.set_syscall("write", check_write, QL_INTERCEPT.ENTER)
|
|
ql.run()
|
|
|
|
self.assertGreaterEqual(len(logged), 2)
|
|
self.assertTrue(logged[-2].startswith('thread 1 ret val is'))
|
|
self.assertTrue(logged[-1].startswith('thread 2 ret val is'))
|
|
|
|
@unittest.skip('broken: unicorn.unicorn.UcError: Invalid instruction (UC_ERR_INSN_INVALID)')
|
|
def test_multithread_elf_linux_armeb(self):
|
|
logged: List[str] = []
|
|
|
|
def check_write(ql: Qiling, fd: int, write_buf, count: int):
|
|
if fd == 1:
|
|
content = ql.mem.read(write_buf, count)
|
|
|
|
logged.extend(content.decode().splitlines())
|
|
|
|
ql = Qiling([fr'{ARMEB_LINUX_ROOTFS}/bin/armeb_multithreading'], ARMEB_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stats = QlOsNullStats()
|
|
ql.os.set_syscall("write", check_write, QL_INTERCEPT.ENTER)
|
|
ql.run()
|
|
|
|
self.assertGreaterEqual(len(logged), 2)
|
|
self.assertTrue(logged[-2].startswith('thread 1 ret val is'))
|
|
self.assertTrue(logged[-1].startswith('thread 2 ret val is'))
|
|
|
|
def test_clone3_translates_to_clone(self):
|
|
# clone3(struct clone_args *, size) must unpack the struct and delegate
|
|
# to the legacy clone() handler with translated args. Modern glibc issues
|
|
# clone3 from pthread_create, so without this thread creation fails.
|
|
# This drives ql_syscall_clone3 directly (stock unicorn, no clone3 binary
|
|
# needed) and asserts the translation.
|
|
CL = 0x100000
|
|
FLAGS, CHILD_TID, PARENT_TID = 0x00010f00, 0x222000, 0x223000
|
|
EXIT_SIGNAL, STACK, STACK_SIZE, TLS = 0x11, 0x7ff00000, 0x8000, 0x224000
|
|
|
|
def run_case(archtype: QL_ARCH, rootfs: str):
|
|
ql = Qiling(code=b"\x00\x00\x00\x00", archtype=archtype, ostype=QL_OS.LINUX,
|
|
rootfs=rootfs, verbose=QL_VERBOSE.OFF)
|
|
ql.mem.map(CL, 0x1000)
|
|
for off, val in ((0, FLAGS), (16, CHILD_TID), (24, PARENT_TID),
|
|
(32, EXIT_SIGNAL), (40, STACK), (48, STACK_SIZE), (56, TLS)):
|
|
ql.mem.write_ptr(CL + off, val, 8)
|
|
|
|
captured = {}
|
|
|
|
def fake_clone(ql, flags, child_stack, parent_tidptr, newtls, child_tidptr):
|
|
captured.update(flags=flags, child_stack=child_stack, parent_tidptr=parent_tidptr,
|
|
newtls=newtls, child_tidptr=child_tidptr)
|
|
return 0xabc
|
|
|
|
orig = ql_sched.ql_syscall_clone
|
|
ql_sched.ql_syscall_clone = fake_clone
|
|
try:
|
|
ret = ql_sched.ql_syscall_clone3(ql, CL, 88)
|
|
finally:
|
|
ql_sched.ql_syscall_clone = orig
|
|
|
|
return captured, ret
|
|
|
|
# generic path (no x8664 register swap)
|
|
cap, ret = run_case(QL_ARCH.ARM, fr'{ARM_LINUX_ROOTFS}')
|
|
self.assertEqual(ret, 0xabc)
|
|
self.assertEqual(cap['child_stack'], STACK + STACK_SIZE) # base + size -> stack top
|
|
self.assertEqual(cap['flags'], FLAGS | EXIT_SIGNAL) # exit_signal folded in
|
|
self.assertEqual(cap['parent_tidptr'], PARENT_TID)
|
|
self.assertEqual(cap['newtls'], TLS)
|
|
self.assertEqual(cap['child_tidptr'], CHILD_TID)
|
|
|
|
# x8664: clone3 pre-swaps newtls<->child_tid so ql_syscall_clone's own
|
|
# x8664 swap cancels out to the same logical mapping
|
|
cap, _ = run_case(QL_ARCH.X8664, fr'{X64_LINUX_ROOTFS}')
|
|
self.assertEqual(cap['newtls'], CHILD_TID)
|
|
self.assertEqual(cap['child_tidptr'], TLS)
|
|
|
|
def test_tcp_elf_linux_x86(self):
|
|
logged: List[str] = []
|
|
|
|
def check_write(ql: Qiling, fd: int, write_buf, count: int):
|
|
if fd == 2:
|
|
content = ql.mem.read(write_buf, count)
|
|
|
|
logged.extend(content.decode().splitlines())
|
|
|
|
ql = Qiling([fr'{X86_LINUX_ROOTFS}/bin/x86_tcp_test', '20000'], X86_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stats = QlOsNullStats()
|
|
ql.os.set_syscall("write", check_write, QL_INTERCEPT.ENTER)
|
|
ql.run()
|
|
|
|
self.assertGreaterEqual(len(logged), 2)
|
|
self.assertTrue(logged[-2].startswith('server recv()'))
|
|
self.assertTrue(logged[-1].startswith('server send()'))
|
|
|
|
# the server is expected to send the value it received, for example:
|
|
# 'server recv() return 14.\n'
|
|
# 'server send() 14 return 14.\n'
|
|
|
|
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
|
|
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
|
|
|
|
num = m.group('num')
|
|
msg = logged[-1].strip()
|
|
|
|
self.assertTrue(msg.endswith(f'{num} return {num}.'))
|
|
|
|
def test_tcp_elf_linux_x8664(self):
|
|
logged: List[str] = []
|
|
|
|
def check_write(ql: Qiling, fd: int, write_buf, count: int):
|
|
if fd == 2:
|
|
content = ql.mem.read(write_buf, count)
|
|
|
|
logged.extend(content.decode().splitlines())
|
|
|
|
ql = Qiling([fr'{X64_LINUX_ROOTFS}/bin/x8664_tcp_test', '20001'], X64_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stats = QlOsNullStats()
|
|
ql.os.set_syscall("write", check_write, QL_INTERCEPT.ENTER)
|
|
ql.run()
|
|
|
|
self.assertGreaterEqual(len(logged), 2)
|
|
self.assertTrue(logged[-2].startswith('server recv()'))
|
|
self.assertTrue(logged[-1].startswith('server send()'))
|
|
|
|
# the server is expected to send the value it received, for example:
|
|
# 'server recv() return 14.\n'
|
|
# 'server send() 14 return 14.\n'
|
|
|
|
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
|
|
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
|
|
|
|
num = m.group('num')
|
|
msg = logged[-1].strip()
|
|
|
|
self.assertTrue(msg.endswith(f'{num} return {num}.'))
|
|
|
|
def test_tcp_elf_linux_arm(self):
|
|
logged: List[str] = []
|
|
|
|
def check_write(ql: Qiling, fd: int, write_buf, count: int):
|
|
if fd == 2:
|
|
content = ql.mem.read(write_buf, count)
|
|
|
|
logged.extend(content.decode().splitlines())
|
|
|
|
ql = Qiling([fr'{ARM_LINUX_ROOTFS}/bin/arm_tcp_test', '20002'], ARM_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stats = QlOsNullStats()
|
|
ql.os.set_syscall("write", check_write, QL_INTERCEPT.ENTER)
|
|
ql.run()
|
|
|
|
self.assertGreaterEqual(len(logged), 2)
|
|
self.assertTrue(logged[-2].startswith('server read()'))
|
|
self.assertTrue(logged[-1].startswith('server write()'))
|
|
|
|
# the server is expected to send the value it received, for example:
|
|
# 'server read() return 14.\n'
|
|
# 'server write() 14 return 14.\n'
|
|
|
|
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
|
|
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
|
|
|
|
num = m.group('num')
|
|
msg = logged[-1].strip()
|
|
|
|
self.assertTrue(msg.endswith(f'{num} return {num}.'))
|
|
|
|
def test_tcp_elf_linux_arm64(self):
|
|
logged: List[str] = []
|
|
|
|
def check_write(ql: Qiling, fd: int, write_buf, count: int):
|
|
if fd == 2:
|
|
content = ql.mem.read(write_buf, count)
|
|
|
|
logged.extend(content.decode().splitlines())
|
|
|
|
ql = Qiling([fr'{ARM64_LINUX_ROOTFS}/bin/arm64_tcp_test', '20003'], ARM64_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stats = QlOsNullStats()
|
|
ql.os.set_syscall("write", check_write, QL_INTERCEPT.ENTER)
|
|
ql.run()
|
|
|
|
self.assertGreaterEqual(len(logged), 2)
|
|
self.assertTrue(logged[-2].startswith('server recv()'))
|
|
self.assertTrue(logged[-1].startswith('server send()'))
|
|
|
|
# the server is expected to send the value it received, for example:
|
|
# 'server recv() return 14.\n'
|
|
# 'server send() 14 return 14.\n'
|
|
|
|
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
|
|
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
|
|
|
|
num = m.group('num')
|
|
msg = logged[-1].strip()
|
|
|
|
self.assertTrue(msg.endswith(f'{num} return {num}.'))
|
|
|
|
def test_tcp_elf_linux_armeb(self):
|
|
logged: List[str] = []
|
|
|
|
def check_write(ql: Qiling, fd: int, write_buf, count: int):
|
|
if fd == 2:
|
|
content = ql.mem.read(write_buf, count)
|
|
|
|
logged.extend(content.decode().splitlines())
|
|
|
|
ql = Qiling([fr'{ARMEB_LINUX_ROOTFS}/bin/armeb_tcp_test', '20004'], ARMEB_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stats = QlOsNullStats()
|
|
ql.os.set_syscall("write", check_write, QL_INTERCEPT.ENTER)
|
|
ql.run()
|
|
|
|
self.assertGreaterEqual(len(logged), 2)
|
|
self.assertTrue(logged[-2].startswith('server recv()'))
|
|
self.assertTrue(logged[-1].startswith('server send()'))
|
|
|
|
# the server is expected to send the value it received, for example:
|
|
# 'server recv() return 14.\n'
|
|
# 'server send() 14 return 14.\n'
|
|
|
|
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
|
|
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
|
|
|
|
num = m.group('num')
|
|
msg = logged[-1].strip()
|
|
|
|
self.assertTrue(msg.endswith(f'{num} return {num}.'))
|
|
|
|
def test_tcp_elf_linux_mips32eb(self):
|
|
logged: List[str] = []
|
|
|
|
def check_write(ql: Qiling, fd: int, write_buf, count: int):
|
|
if fd == 2:
|
|
content = ql.mem.read(write_buf, count)
|
|
|
|
logged.extend(content.decode().splitlines())
|
|
|
|
ql = Qiling([fr'{MIPSEB_LINUX_ROOTFS}/bin/mips32_tcp_test', '20005'], MIPSEB_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stats = QlOsNullStats()
|
|
ql.os.set_syscall("write", check_write, QL_INTERCEPT.ENTER)
|
|
ql.run()
|
|
|
|
self.assertGreaterEqual(len(logged), 2)
|
|
self.assertTrue(logged[-2].startswith('server recv()'))
|
|
self.assertTrue(logged[-1].startswith('server send()'))
|
|
|
|
# the server is expected to send the value it received, for example:
|
|
# 'server recv() return 14.\n'
|
|
# 'server send() 14 return 14.\n'
|
|
|
|
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
|
|
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
|
|
|
|
num = m.group('num')
|
|
msg = logged[-1].strip()
|
|
|
|
self.assertTrue(msg.endswith(f'{num} return {num}.'))
|
|
|
|
def test_tcp_elf_linux_mips32el(self):
|
|
logged: List[str] = []
|
|
|
|
def check_write(ql: Qiling, fd: int, write_buf, count: int):
|
|
if fd == 2:
|
|
content = ql.mem.read(write_buf, count)
|
|
|
|
logged.extend(content.decode().splitlines())
|
|
|
|
ql = Qiling([fr'{MIPSEL_LINUX_ROOTFS}/bin/mips32el_tcp_test', '20006'], MIPSEL_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stats = QlOsNullStats()
|
|
ql.os.set_syscall("write", check_write, QL_INTERCEPT.ENTER)
|
|
ql.run()
|
|
|
|
self.assertGreaterEqual(len(logged), 2)
|
|
self.assertTrue(logged[-2].startswith('server read()'))
|
|
self.assertTrue(logged[-1].startswith('server write()'))
|
|
|
|
# the server is expected to send the value it received, for example:
|
|
# 'server read() return 14.\n'
|
|
# 'server write() 14 return 14.\n'
|
|
|
|
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
|
|
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
|
|
|
|
num = m.group('num')
|
|
msg = logged[-1].strip()
|
|
|
|
self.assertTrue(msg.endswith(f'{num} return {num}.'))
|
|
|
|
def test_udp_elf_linux_x86(self):
|
|
logged: List[str] = []
|
|
|
|
def check_write(ql: Qiling, fd: int, write_buf, count: int):
|
|
if fd == 2:
|
|
content = ql.mem.read(write_buf, count)
|
|
|
|
logged.extend(content.decode().splitlines())
|
|
|
|
ql = Qiling([fr'{X86_LINUX_ROOTFS}/bin/x86_udp_test', '20010'], X86_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stats = QlOsNullStats()
|
|
ql.os.set_syscall("write", check_write, QL_INTERCEPT.ENTER)
|
|
ql.run()
|
|
|
|
self.assertGreaterEqual(len(logged), 2)
|
|
self.assertTrue(logged[-2].startswith('server recvfrom()'))
|
|
self.assertTrue(logged[-1].startswith('server sendto()'))
|
|
|
|
# the server is expected to send the value it received, for example:
|
|
# 'server recvfrom() return 14.\n'
|
|
# 'server sendto() 14 return 14.\n'
|
|
|
|
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
|
|
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
|
|
|
|
num = m.group('num')
|
|
msg = logged[-1].strip()
|
|
|
|
self.assertTrue(msg.endswith(f'{num} return {num}.'))
|
|
|
|
def test_udp_elf_linux_x8664(self):
|
|
logged: List[str] = []
|
|
|
|
def check_write(ql: Qiling, fd: int, write_buf, count: int):
|
|
if fd == 2:
|
|
content = ql.mem.read(write_buf, count)
|
|
|
|
logged.extend(content.decode().splitlines())
|
|
|
|
ql = Qiling([fr'{X64_LINUX_ROOTFS}/bin/x8664_udp_test', '20011'], X64_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stats = QlOsNullStats()
|
|
ql.os.set_syscall("write", check_write, QL_INTERCEPT.ENTER)
|
|
ql.run()
|
|
|
|
self.assertGreaterEqual(len(logged), 2)
|
|
self.assertTrue(logged[-2].startswith('server recvfrom()'))
|
|
self.assertTrue(logged[-1].startswith('server sendto()'))
|
|
|
|
# the server is expected to send the value it received, for example:
|
|
# 'server recvfrom() return 14.\n'
|
|
# 'server sendto() 14 return 14.\n'
|
|
|
|
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
|
|
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
|
|
|
|
num = m.group('num')
|
|
msg = logged[-1].strip()
|
|
|
|
self.assertTrue(msg.endswith(f'{num} return {num}.'))
|
|
|
|
def test_udp_elf_linux_arm64(self):
|
|
logged: List[str] = []
|
|
|
|
def check_write(ql: Qiling, fd: int, write_buf, count: int):
|
|
if fd == 2:
|
|
content = ql.mem.read(write_buf, count)
|
|
|
|
logged.extend(content.decode().splitlines())
|
|
|
|
ql = Qiling([fr'{ARM64_LINUX_ROOTFS}/bin/arm64_udp_test', '20013'], ARM64_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stats = QlOsNullStats()
|
|
ql.os.set_syscall("write", check_write, QL_INTERCEPT.ENTER)
|
|
ql.run()
|
|
|
|
self.assertGreaterEqual(len(logged), 2)
|
|
self.assertTrue(logged[-2].startswith('server recvfrom()'))
|
|
self.assertTrue(logged[-1].startswith('server sendto()'))
|
|
|
|
# the server is expected to send the value it received, for example:
|
|
# 'server recvfrom() return 14.\n'
|
|
# 'server sendto() 14 return 14.\n'
|
|
|
|
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
|
|
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
|
|
|
|
num = m.group('num')
|
|
msg = logged[-1].strip()
|
|
|
|
self.assertTrue(msg.endswith(f'{num} return {num}.'))
|
|
|
|
def test_udp_elf_linux_armeb(self):
|
|
logged: List[str] = []
|
|
|
|
def check_write(ql: Qiling, fd: int, write_buf, count: int):
|
|
if fd == 2:
|
|
content = ql.mem.read(write_buf, count)
|
|
|
|
logged.extend(content.decode().splitlines())
|
|
|
|
ql = Qiling([fr'{ARMEB_LINUX_ROOTFS}/bin/armeb_udp_test', '20014'], ARMEB_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
|
|
ql.os.stats = QlOsNullStats()
|
|
ql.os.set_syscall("write", check_write, QL_INTERCEPT.ENTER)
|
|
ql.run()
|
|
|
|
self.assertGreaterEqual(len(logged), 2)
|
|
self.assertTrue(logged[-2].startswith('server recvfrom()'))
|
|
self.assertTrue(logged[-1].startswith('server sendto()'))
|
|
|
|
# the server is expected to send the value it received, for example:
|
|
# 'server recvfrom() return 14.\n'
|
|
# 'server sendto() 14 return 14.\n'
|
|
|
|
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
|
|
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
|
|
|
|
num = m.group('num')
|
|
msg = logged[-1].strip()
|
|
|
|
self.assertTrue(msg.endswith(f'{num} return {num}.'))
|
|
|
|
def test_http_elf_linux_x8664(self):
|
|
PORT = 20020
|
|
|
|
def picohttpd():
|
|
ql = Qiling([fr'{X64_LINUX_ROOTFS}/bin/picohttpd', f'{PORT:d}'], X64_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
ql.run()
|
|
|
|
picohttpd_therad = threading.Thread(target=picohttpd, daemon=True)
|
|
picohttpd_therad.start()
|
|
|
|
time.sleep(1)
|
|
|
|
conn = http.client.HTTPConnection('localhost', PORT, timeout=10)
|
|
conn.request('GET', '/')
|
|
|
|
# libc uses statx to query stdout stats, but fails because 'stdout' is not a valid path
|
|
# on the hosting paltform. it prints out the "Server started" message, but stdout is not
|
|
# found and the message is kept buffered in.
|
|
#
|
|
# later on, picohttpd dups the client socket into stdout fd and uses ordinary printf to
|
|
# send data out. however, when the "successful" message is sent, it is sent along with
|
|
# the buffered message, which arrives first and raises a http.client.BadStatusLine exception
|
|
# as it reads as a malformed http response.
|
|
#
|
|
# here we first peek at the incoming buffer to see whether we got a surplus prefix message.
|
|
# in case we have, we discard it first and only then read the http response
|
|
|
|
# look for the http response header
|
|
incoming = conn.sock.recv(96, socket.MSG_PEEK)
|
|
surplus = incoming.find(b'HTTP/1.1')
|
|
|
|
# if incoming buffer has a surplus prefix, discard it
|
|
if surplus > 0:
|
|
conn.sock.recv(surplus)
|
|
|
|
response = conn.getresponse()
|
|
feedback = response.read()
|
|
self.assertEqual('httpd_test_successful', feedback.decode())
|
|
|
|
def test_http_elf_linux_arm(self):
|
|
PORT = 20021
|
|
|
|
def picohttpd():
|
|
ql = Qiling([fr'{ARM_LINUX_ROOTFS}/bin/picohttpd', f'{PORT:d}'], ARM_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
ql.run()
|
|
|
|
picohttpd_therad = threading.Thread(target=picohttpd, daemon=True)
|
|
picohttpd_therad.start()
|
|
|
|
time.sleep(1)
|
|
|
|
conn = http.client.HTTPConnection('localhost', PORT, timeout=10)
|
|
conn.request('GET', '/')
|
|
|
|
# look for the http response header
|
|
incoming = conn.sock.recv(96, socket.MSG_PEEK)
|
|
surplus = incoming.find(b'HTTP/1.1')
|
|
|
|
# if incoming buffer has a surplus prefix, discard it
|
|
if surplus > 0:
|
|
conn.sock.recv(surplus)
|
|
|
|
response = conn.getresponse()
|
|
feedback = response.read()
|
|
self.assertEqual('httpd_test_successful', feedback.decode())
|
|
|
|
def test_http_elf_linux_armeb(self):
|
|
PORT = 20022
|
|
|
|
def picohttpd():
|
|
ql = Qiling([fr'{ARMEB_LINUX_ROOTFS}/bin/picohttpd', f'{PORT:d}'], ARMEB_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
|
|
ql.run()
|
|
|
|
picohttpd_thread = threading.Thread(target=picohttpd, daemon=True)
|
|
picohttpd_thread.start()
|
|
|
|
time.sleep(1)
|
|
|
|
conn = http.client.HTTPConnection('localhost', PORT, timeout=10)
|
|
conn.request('GET', '/')
|
|
|
|
# look for the http response header
|
|
incoming = conn.sock.recv(96, socket.MSG_PEEK)
|
|
surplus = incoming.find(b'HTTP/1.1')
|
|
|
|
# if incoming buffer has a surplus prefix, discard it
|
|
if surplus > 0:
|
|
conn.sock.recv(surplus)
|
|
|
|
response = conn.getresponse()
|
|
feedback = response.read()
|
|
self.assertEqual('httpd_test_successful', feedback.decode())
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|