Merge branch 'qilingframework:dev' into dev

This commit is contained in:
libumem
2025-11-12 20:23:27 -05:00
committed by GitHub
63 changed files with 1667 additions and 719 deletions
+97 -101
View File
@@ -1,4 +1,4 @@
[![Documentation Status](https://readthedocs.org/projects/qilingframework/badge/?version=latest)](https://docs.qiling.io)
[![Documentation Status](https://github.com/qilingframework/qiling/wiki)](https://github.com/qilingframework/qiling/wiki)
[![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,24 +8,43 @@
<img width="150" height="150" src="https://raw.githubusercontent.com/qilingframework/qiling/master/docs/qiling2_logo_small.png">
</p>
[Qiling's use case, blog and related work](https://github.com/qilingframework/qiling/issues/134)
# Qiling Framework
Qiling is an advanced binary emulation framework, with the following features:
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.
- 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.
## 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
Qiling also made its way to various international conferences.
@@ -49,79 +68,37 @@ 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 is backed by [Unicorn Engine](http://www.unicorn-engine.org).
Qiling has been presented at various international conferences, showcasing its versatility in:
Visit our [website](https://www.qiling.io) for more information.
- Binary analysis and reverse engineering.
- Malware analysis and sandboxing.
- Firmware analysis and emulation.
- Security research and vulnerability discovery.
- CTF challenges and exploit development.
---
#### License
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).
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.
## Quick Start
---
### Installation
#### Qiling vs. other Emulators
Qiling requires Python 3.8 or newer. You can install it using pip:
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.
```bash
pip install qiling
```
##### Qiling vs. Unicorn engine
For more detailed installation instructions and dependencies, please refer to the [official documentation](https://github.com/qilingframework/qiling/wiki/Installation).
Built on top of Unicorn, but Qiling & Unicorn are two different animals.
### Basic Usage
- 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://docs.qiling.io/en/latest/install/) 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.
The example below shows how to use Qiling framework in the most straightforward way to emulate a Windows executable.
```python
from qiling import Qiling
@@ -135,8 +112,30 @@ if __name__ == "__main__":
ql.run()
```
- The following example shows how a Windows crackme may be patched dynamically
to make it always display the “Congratulation” dialog.
## 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.
```python
from qiling import Qiling
@@ -177,15 +176,13 @@ 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_T3KLh2NU)
[![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'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)
@@ -195,63 +192,62 @@ 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
#### Contact
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.
Get the latest info from our website https://www.qiling.io
## License
Contact us at email info@qiling.io,
via Twitter [@qiling_io](https://twitter.com/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
#### Core developers, Key Contributors and etc.
Get the latest info from our website [https://www.qiling.io](https://www.qiling.io)
Please refer to [CREDITS.md](https://github.com/qilingframework/qiling/blob/dev/CREDITS.md).
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).
+4
View File
@@ -0,0 +1,4 @@
[CODE]
load_address = 0x10000000
entry_point = 0x10000008
ram_size = 0xa00000
@@ -5,7 +5,7 @@
# Everything about the bug and firmware https://www.exploit-db.com/exploits/33863
import os,sys
import sys
sys.path.append("../../..")
from qiling import Qiling
@@ -13,7 +13,7 @@ from qiling.const import QL_VERBOSE
from qiling.extensions.afl import ql_afl_fuzz
def main(input_file, enable_trace=False):
def main(input_file: str):
env_vars = {
"REQUEST_METHOD": "POST",
@@ -24,40 +24,36 @@ def main(input_file, enable_trace=False):
# "CONTENT_LENGTH": "8", # no needed
}
ql = Qiling(["./rootfs/htdocs/web/hedwig.cgi"], "./rootfs",
verbose=QL_VERBOSE.DEBUG, env=env_vars, console=enable_trace)
ql = Qiling(["./rootfs/htdocs/web/hedwig.cgi"], "./rootfs", verbose=QL_VERBOSE.DISABLED, env=env_vars)
def place_input_callback(ql: Qiling, input: bytes, _: int):
env_var = ("HTTP_COOKIE=uid=1234&password=").encode()
env_vars = env_var + input + b"\x00" + (ql.path).encode() + b"\x00"
ql.mem.write(ql.target_addr, env_vars)
def place_input_callback(ql: Qiling, data: bytes, _: int) -> bool:
# construct the payload
payload = b''.join((b"HTTP_COOKIE=uid=1234&password=", bytes(data), b"\x00", ql_path, b"\x00"))
# patch the value of 'HTTP_COOKIE' in memory
ql.mem.write(target_addr, payload)
# payload is in place, we are good to go
return True
def start_afl(_ql: Qiling):
"""
Callback from inside
"""
ql_afl_fuzz(_ql, input_file=input_file, place_input_callback=place_input_callback, exits=[ql.os.exit_point])
addr = ql.mem.search("HTTP_COOKIE=uid=1234&password=".encode())
ql.target_addr = addr[0]
addr = ql.mem.search(b"HTTP_COOKIE=uid=1234&password=")
target_addr = addr[0]
ql_path = ql.path.encode()
main_addr = ql.loader.elf_entry
ql.hook_address(callback=start_afl, address=main_addr)
ql.hook_address(start_afl, ql.loader.elf_entry)
try:
ql.run()
os._exit(0)
except:
if enable_trace:
print("\nFuzzer Went Shit")
os._exit(0)
ql.run()
if __name__ == "__main__":
if len(sys.argv) == 1:
if len(sys.argv) < 2:
raise ValueError("No input file provided.")
if len(sys.argv) > 2 and sys.argv[1] == "-t":
main(sys.argv[2], enable_trace=True)
else:
main(sys.argv[1])
main(sys.argv[1])
+101
View File
@@ -0,0 +1,101 @@
##############################################################################
# This example is meant to demonstrate the modifications necessary
# to enable code coverage when emulating small code snippets or bare-metal
# code.
##############################################################################
from qiling import Qiling
from qiling.const import QL_ARCH, QL_OS, QL_VERBOSE
from qiling.extensions.coverage import utils as cov_utils
from qiling.loader.loader import Image
import os
BASE_ADDRESS = 0x10000000
CHECKSUM_FUNC_ADDR = BASE_ADDRESS + 0x8
END_ADDRESS = 0x100000ba
DATA_ADDR = 0xa0000000 # Arbitrary address for data
STACK_ADDR = 0xb0000000 # Arbitrary address for stack
# Python implementation of the checksum function being emulated
# This checksum function is intended to have different code paths based on the input
# which is useful for observing code coverage
def checksum_function(input_data_buffer: bytes):
expected_checksum_python = 0
input_data_len = len(input_data_buffer)
if input_data_len >= 1 and input_data_buffer[0] == 0xDE: # MAGIC_VALUE_1
for i in range(min(input_data_len, 4)):
expected_checksum_python += input_data_buffer[i]
expected_checksum_python += 0x10
elif input_data_len >= 2 and input_data_buffer[1] == 0xAD: # MAGIC_VALUE_2
for i in range(input_data_len):
expected_checksum_python ^= input_data_buffer[i]
expected_checksum_python += 0x20
else:
for i in range(input_data_len):
expected_checksum_python += input_data_buffer[i]
expected_checksum_python &= 0xFF # Ensure it's a single byte
return expected_checksum_python
def unmapped_handler(ql: Qiling, type: int, addr: int, size: int, value: int) -> None:
print(f"Unmapped Memory R/W, trying to access {size:d} bytes at {addr:#010x} from {ql.arch.regs.pc:#010x}")
def emulate_checksum_function(input_data_buffer: bytes) -> None:
print(f"\n--- Testing with input: {input_data_buffer.hex()} ---")
test_file = "rootfs/blob/example_raw.bin"
with open(test_file, "rb") as f:
raw_code: bytes = f.read()
ql: Qiling = Qiling(
code=raw_code,
archtype=QL_ARCH.ARM,
ostype=QL_OS.BLOB,
profile="blob_raw.ql",
verbose=QL_VERBOSE.DEBUG,
thumb=True
)
''' monkeypatch - Correcting the loader image name, used for coverage collection
removing all images with name 'blob_code' that were created by the blob loader.
This is necessary because some code coverage visualization tools require the
module name to match that of the input file '''
ql.loader.images = [img for img in ql.loader.images if img.path != 'blob_code']
ql.loader.images.append(Image(ql.loader.load_address, ql.loader.load_address + ql.os.code_ram_size, os.path.basename(test_file)))
input_data_len: int = len(input_data_buffer)
# Map memory for the data and stack
ql.mem.map(STACK_ADDR, 0x2000)
ql.mem.map(DATA_ADDR, ql.mem.align_up(input_data_len + 0x100)) # Map enough space for data
# Write input data
ql.mem.write(DATA_ADDR, input_data_buffer)
# Set up the stack pointer
ql.arch.regs.sp = STACK_ADDR + 0x2000 - 4
# Set up argument registers
ql.arch.regs.r0 = DATA_ADDR
ql.arch.regs.r1 = input_data_len
# Set the program counter to the function's entry point
ql.arch.regs.pc = CHECKSUM_FUNC_ADDR
# Set the return address (LR) to a dummy address.
ql.arch.regs.lr = 0xbebebebe
ql.hook_mem_unmapped(unmapped_handler)
#ql.debugger="gdb:127.0.0.1:9999"
# Start emulation
print(f"Starting emulation at PC: {hex(ql.arch.regs.pc)}")
try:
with cov_utils.collect_coverage(ql, 'drcov', 'output.cov'):
ql.run(begin=CHECKSUM_FUNC_ADDR, end=END_ADDRESS)
except Exception as e:
print(f"Emulation error: {e}")
print(f"Emulated checksum: {hex(ql.arch.regs.r0)}")
if __name__ == "__main__":
data = b"\x01\x02\x03\x04\x05" # Example input data
emulate_checksum_function(data)
+59 -45
View File
@@ -8,68 +8,82 @@ sys.path.append("..")
from qiling.core import Qiling
from qiling.const import QL_ARCH, QL_OS, QL_VERBOSE
from qiling.os.const import STRING
from qiling.os.const import STRING, SIZE_T, POINTER
def get_kaimendaji_password():
def my_getenv(ql: Qiling):
env = {
"ID" : b"000000000000000",
"ethaddr" : b"11:22:33:44:55:66"
}
def my_getenv(ql: Qiling):
env = {
"ID" : b"000000000000000",
"ethaddr" : b"11:22:33:44:55:66"
}
params = ql.os.resolve_fcall_params({'key': STRING})
value = env.get(params["key"], b"")
params = ql.os.resolve_fcall_params({'key': STRING})
value = env.get(params["key"], b"")
value_addr = ql.os.heap.alloc(len(value))
ql.mem.write(value_addr, value)
value_addr = ql.os.heap.alloc(len(value))
ql.mem.write(value_addr, value)
ql.arch.regs.r0 = value_addr
ql.arch.regs.arch_pc = ql.arch.regs.lr
ql.arch.regs.r0 = value_addr
ql.arch.regs.arch_pc = ql.arch.regs.lr
def get_password(ql: Qiling):
password_raw = ql.mem.read(ql.arch.regs.r0, ql.arch.regs.r2)
password = ''
for item in password_raw:
if 0 <= item <= 9:
password += chr(item + 48)
else:
password += chr(item + 87)
def get_password(ql: Qiling):
# we land on a memcmp call, where the real password is being compared to
# the one provided by the user. we can follow the arguments to read the
# real password
print("The password is: %s" % password)
params = ql.os.resolve_fcall_params({
'ptr1': POINTER, # points to real password
'ptr2': POINTER, # points to user provided password
'size': SIZE_T # comparison length
})
def partial_run_init(ql: Qiling):
# argv prepare
ql.arch.regs.arch_sp -= 0x30
arg0_ptr = ql.arch.regs.arch_sp
ql.mem.write(arg0_ptr, b"kaimendaji")
ptr1 = params['ptr1']
size = params['size']
ql.arch.regs.arch_sp -= 0x10
arg1_ptr = ql.arch.regs.arch_sp
ql.mem.write(arg1_ptr, b"000000") # arbitrary password
password_raw = ql.mem.read(ptr1, size)
ql.arch.regs.arch_sp -= 0x20
argv_ptr = ql.arch.regs.arch_sp
ql.mem.write_ptr(argv_ptr, arg0_ptr)
ql.mem.write_ptr(argv_ptr + ql.arch.pointersize, arg1_ptr)
def __hex_digit(ch: int) -> str:
off = ord('0') if ch in range(10) else ord('a') - 10
ql.arch.regs.r2 = 2
ql.arch.regs.r3 = argv_ptr
return chr(ch + off)
with open("../examples/rootfs/blob/u-boot.bin.img", "rb") as f:
uboot_code = f.read()
# should be: "013f1f"
password = "".join(__hex_digit(ch) for ch in password_raw)
ql = Qiling(code=uboot_code[0x40:], archtype=QL_ARCH.ARM, ostype=QL_OS.BLOB, profile="uboot_bin.ql", verbose=QL_VERBOSE.OFF)
print(f'The password is: {password}')
image_base_addr = ql.loader.load_address
ql.hook_address(my_getenv, image_base_addr + 0x13AC0)
ql.hook_address(get_password, image_base_addr + 0x48634)
partial_run_init(ql)
def partial_run_init(ql: Qiling):
# argv prepare
ql.arch.regs.arch_sp -= 0x30
arg0_ptr = ql.arch.regs.arch_sp
ql.mem.write(arg0_ptr, b"kaimendaji")
ql.run(image_base_addr + 0x486B4, image_base_addr + 0x48718)
ql.arch.regs.arch_sp -= 0x10
arg1_ptr = ql.arch.regs.arch_sp
ql.mem.write(arg1_ptr, b"000000") # arbitrary password
ql.arch.regs.arch_sp -= 0x20
argv_ptr = ql.arch.regs.arch_sp
ql.mem.write_ptr(argv_ptr, arg0_ptr)
ql.mem.write_ptr(argv_ptr + ql.arch.pointersize, arg1_ptr)
ql.arch.regs.r2 = 2
ql.arch.regs.r3 = argv_ptr
if __name__ == "__main__":
get_kaimendaji_password()
with open("../examples/rootfs/blob/u-boot.bin.img", "rb") as f:
uboot_code = f.read()
ql = Qiling(code=uboot_code[0x40:], archtype=QL_ARCH.ARM, ostype=QL_OS.BLOB, profile="uboot_bin.ql", verbose=QL_VERBOSE.DEBUG)
imgbase = ql.loader.images[0].base
ql.hook_address(my_getenv, imgbase + 0x13AC0)
ql.hook_address(get_password, imgbase + 0x48634)
partial_run_init(ql)
ql.run(imgbase + 0x486B4, imgbase + 0x48718)
+1 -1
View File
@@ -159,7 +159,7 @@ def hook_StartServiceA(ql: Qiling, address: int, params):
init_unseen_symbols(ql.amsint32_driver, ntoskrnl.base+0xb7695, b"NtTerminateProcess", 0, "ntoskrnl.exe")
#ql.amsint32_driver.debugger= ":9999"
try:
ql.amsint32_driver.load()
ql.amsint32_driver.run()
return 1
except UcError as e:
print("Load driver error: ", e)
+52
View File
@@ -0,0 +1,52 @@
# Makefile for Bare-Metal ARM Checksum Calculator
# --- Toolchain Definitions ---
TOOLCHAIN_PREFIX = arm-none-eabi
# Compiler, Linker, and Objcopy executables
CC = $(TOOLCHAIN_PREFIX)-gcc
LD = $(TOOLCHAIN_PREFIX)-gcc
OBJCOPY = $(TOOLCHAIN_PREFIX)-objcopy
# --- Source and Output Files ---
SRCS = example_raw.c
OBJS = $(SRCS:.c=.o) # Convert .c to .o
ELF = example_raw.elf
BIN = example_raw.bin
# --- Linker Script ---
LDSCRIPT = linker.ld
# --- Compiler Flags ---
CFLAGS = -c -O0 -mcpu=cortex-a7 -mthumb -ffreestanding -nostdlib
# --- Linker Flags ---
LDFLAGS = -T $(LDSCRIPT) -nostdlib
# --- Objcopy Flags ---
OBJCOPYFLAGS = -O binary
# --- Default Target ---
.PHONY: all clean
all: $(BIN)
# Rule to build the raw binary (.bin) from the ELF file
$(BIN): $(ELF)
$(OBJCOPY) $(OBJCOPYFLAGS) $< $@
@echo "Successfully created $(BIN)"
# Rule to link the object file into an ELF executable
$(ELF): $(OBJS) $(LDSCRIPT)
$(LD) $(LDFLAGS) $(OBJS) -o $@
@echo "Successfully linked $(ELF)"
# Rule to compile the C source file into an object file
%.o: %.c
$(CC) $(CFLAGS) $< -o $@
@echo "Successfully compiled $<"
# --- Clean Rule ---
clean:
rm -f $(OBJS) $(ELF) $(BIN)
@echo "Cleaned build artifacts."
+56
View File
@@ -0,0 +1,56 @@
// example checksum algorithm to demonstrate raw binary code coverage in qiling
// example_raw.c
// Define some magic values
#define MAGIC_VALUE_1 0xDE
#define MAGIC_VALUE_2 0xAD
// This function calculates a checksum with branches based on input data
// It takes a pointer to data and its length
// Returns the checksum (unsigned char to fit in a byte)
unsigned char calculate_checksum(const unsigned char *data, unsigned int length) {
unsigned char checksum = 0;
// Branch 1: Check for MAGIC_VALUE_1 at the start
if (length >= 1 && data[0] == MAGIC_VALUE_1) {
// If first byte is MAGIC_VALUE_1, do a simple sum of first 4 bytes
// (or up to length if less than 4)
for (unsigned int i = 0; i < length && i < 4; i++) {
checksum += data[i];
}
// Add a fixed offset to make this path distinct
checksum += 0x10;
}
// Branch 2: Check for MAGIC_VALUE_2 at the second byte
else if (length >= 2 && data[1] == MAGIC_VALUE_2) {
// If second byte is MAGIC_VALUE_2, do a XOR sum of all bytes
for (unsigned int i = 0; i < length; i++) {
checksum ^= data[i];
}
// Add a fixed offset to make this path distinct
checksum += 0x20;
}
// Default Branch: Standard byte sum checksum
else {
for (unsigned int i = 0; i < length; i++) {
checksum += data[i];
}
}
return checksum;
}
// Minimal entry point for bare-metal.
// This function will not be called directly during Qiling emulation,
// but it's needed for the linker to have an entry point.
__attribute__((section(".text.startup")))
void _start() {
// In a real bare-metal application, this would initialize hardware,
// set up stacks, etc. For this example, it's just a placeholder.
// We'll call calculate_checksum directly from our Qiling script.
while (1) {
// Do nothing, or perhaps put the CPU to sleep
asm volatile ("wfi"); // Wait For Interrupt (ARM instruction)
}
}
+39
View File
@@ -0,0 +1,39 @@
/* linker.ld */
ENTRY(_start) /* Define the entry point of our program */
/* Define memory regions - simple RAM region for this example */
MEMORY
{
ram (rwx) : ORIGIN = 0x10000000, LENGTH = 64K /* 64KB of RAM for our program */
}
SECTIONS
{
/* Define the start of our program in memory.
*/
. = 0x10000000;
.text : {
KEEP(*(.text.startup)) /* Keep the _start function */
*(.text) /* All other code */
*(.text.*)
*(.rodata) /* Read-only data */
*(.rodata.*)
. = ALIGN(4);
} > ram /* Place .text section in the 'ram' region */
.data : {
. = ALIGN(4);
*(.data) /* Initialized data */
*(.data.*)
. = ALIGN(4);
} > ram
.bss : {
. = ALIGN(4);
*(.bss)
*(.bss.*)
. = ALIGN(4);
} > ram
}
+2
View File
@@ -78,6 +78,8 @@ def my_sandbox(path, rootfs):
ql.os.set_syscall('vfork', __vfork)
os.unlink(fr'{ROOTFS}/proc/sys/kernel/core_pattern')
ql.run()
+2
View File
@@ -1,6 +1,8 @@
[CODE]
ram_size = 0xa00000
load_address = 0x80800000
entry_point = 0x80800000
heap_address = 0xa0000000
heap_size = 0x300000
+7 -4
View File
@@ -48,7 +48,7 @@ class QlArchUtils:
return addr, '-'
def disassembler(self, ql: Qiling, address: int, size: int):
data = ql.mem.read(address, size)
data = memoryview(ql.mem.read(address, size))
# knowing that all binary sections are aligned to page boundary allows
# us to 'cheat' and search for the containing image using the aligned
@@ -64,11 +64,14 @@ class QlArchUtils:
ba, name = self.get_base_and_name(ql.mem.align(address))
anibbles = ql.arch.bits // 4
pos = 0
for insn in ql.arch.disassembler.disasm(data, address):
offset = insn.address - ba
for iaddr, isize, mnem, ops in ql.arch.disassembler.disasm_lite(data, address):
offset = iaddr - ba
ibytes = data[pos:pos + isize]
ql.log.info(f'{insn.address:0{anibbles}x} [{name:20s} + {offset:#08x}] {insn.bytes.hex(" "):20s} {insn.mnemonic:20s} {insn.op_str}')
ql.log.info(f'{iaddr:0{anibbles}x} [{name:20s} + {offset:#08x}] {ibytes.hex():22s} {mnem:16s} {ops}')
pos += isize
if ql.verbose >= QL_VERBOSE.DUMP:
for reg in ql.arch.regs.register_mapping:
+6
View File
@@ -70,6 +70,12 @@ class QlCC:
raise NotImplementedError
def getReturnAddress(self) -> int:
"""Get function return address.
"""
raise NotImplementedError
def setReturnAddress(self, addr: int) -> None:
"""Set function return address.
+6 -1
View File
@@ -21,17 +21,22 @@ class QlArmBaseCC(QlCommonBaseCC):
def getNumSlots(argbits: int) -> int:
return 1
def getReturnAddress(self) -> int:
return self.arch.regs.lr
def setReturnAddress(self, addr: int) -> None:
self.arch.regs.lr = addr
def unwind(self, nslots: int) -> int:
# TODO: cleanup?
return self.arch.regs.lr
return self.getReturnAddress()
class aarch64(QlArmBaseCC):
_retreg = UC_ARM64_REG_X0
_argregs = make_arg_list(UC_ARM64_REG_X0, UC_ARM64_REG_X1, UC_ARM64_REG_X2, UC_ARM64_REG_X3, UC_ARM64_REG_X4, UC_ARM64_REG_X5, UC_ARM64_REG_X6, UC_ARM64_REG_X7)
class aarch32(QlArmBaseCC):
_retreg = UC_ARM_REG_R0
_argregs = make_arg_list(UC_ARM_REG_R0, UC_ARM_REG_R1, UC_ARM_REG_R2, UC_ARM_REG_R3)
+3
View File
@@ -15,6 +15,9 @@ class QlIntelBaseCC(QlCommonBaseCC):
Supports arguments passing over registers and stack.
"""
def getReturnAddress(self) -> int:
return self.arch.stack_read(0)
def setReturnAddress(self, addr: int) -> None:
self.arch.stack_push(addr)
+3
View File
@@ -12,6 +12,9 @@ class mipso32(QlCommonBaseCC):
_shadow = 4
_retaddr_on_stack = False
def getReturnAddress(self) -> int:
return self.arch.regs.ra
def setReturnAddress(self, addr: int):
self.arch.regs.ra = addr
+3
View File
@@ -22,5 +22,8 @@ class ppc(QlCommonBaseCC):
def getNumSlots(argbits: int):
return 1
def getReturnAddress(self) -> int:
return self.arch.regs.lr
def setReturnAddress(self, addr: int):
self.arch.regs.lr = addr
+3
View File
@@ -22,5 +22,8 @@ class riscv(QlCommonBaseCC):
def getNumSlots(argbits: int):
return 1
def getReturnAddress(self) -> int:
return self.arch.regs.ra
def setReturnAddress(self, addr: int):
self.arch.regs.ra = addr
+25 -25
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
#
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
@@ -25,14 +25,14 @@ class QlCoreStructs:
QL_ENDIAN.EB: '>'
}[endian]
self._fmt8 = f'{modifier}B'
self._fmt8s = f'{modifier}b'
self._fmt16 = f'{modifier}H'
self._fmt16s = f'{modifier}h'
self._fmt32 = f'{modifier}I'
self._fmt32s = f'{modifier}i'
self._fmt64 = f'{modifier}Q'
self._fmt64s = f'{modifier}q'
self._fmt8 = struct.Struct(f'{modifier}B')
self._fmt8s = struct.Struct(f'{modifier}b')
self._fmt16 = struct.Struct(f'{modifier}H')
self._fmt16s = struct.Struct(f'{modifier}h')
self._fmt32 = struct.Struct(f'{modifier}I')
self._fmt32s = struct.Struct(f'{modifier}i')
self._fmt64 = struct.Struct(f'{modifier}Q')
self._fmt64s = struct.Struct(f'{modifier}q')
handlers = {
64 : (self.pack64, self.pack64s, self.unpack64, self.unpack64s),
@@ -51,49 +51,49 @@ class QlCoreStructs:
self.unpacks = ups
def pack64(self, x: int, /) -> bytes:
return struct.pack(self._fmt64, x)
return self._fmt64.pack(x)
def pack64s(self, x: int, /) -> bytes:
return struct.pack(self._fmt64s, x)
return self._fmt64s.pack(x)
def unpack64(self, x: ReadableBuffer, /) -> int:
return struct.unpack(self._fmt64, x)[0]
return self._fmt64.unpack(x)[0]
def unpack64s(self, x: ReadableBuffer, /) -> int:
return struct.unpack(self._fmt64s, x)[0]
return self._fmt64s.unpack(x)[0]
def pack32(self, x: int, /) -> bytes:
return struct.pack(self._fmt32, x)
return self._fmt32.pack(x)
def pack32s(self, x: int, /) -> bytes:
return struct.pack(self._fmt32s, x)
return self._fmt32s.pack(x)
def unpack32(self, x: ReadableBuffer, /) -> int:
return struct.unpack(self._fmt32, x)[0]
return self._fmt32.unpack(x)[0]
def unpack32s(self, x: ReadableBuffer, /) -> int:
return struct.unpack(self._fmt32s, x)[0]
return self._fmt32s.unpack(x)[0]
def pack16(self, x: int, /) -> bytes:
return struct.pack(self._fmt16, x)
return self._fmt16.pack(x)
def pack16s(self, x: int, /) -> bytes:
return struct.pack(self._fmt16s, x)
return self._fmt16s.pack(x)
def unpack16(self, x: ReadableBuffer, /) -> int:
return struct.unpack(self._fmt16, x)[0]
return self._fmt16.unpack(x)[0]
def unpack16s(self, x: ReadableBuffer, /) -> int:
return struct.unpack(self._fmt16s, x)[0]
return self._fmt16s.unpack(x)[0]
def pack8(self, x: int, /) -> bytes:
return struct.pack(self._fmt8, x)
return self._fmt8.pack(x)
def pack8s(self, x: int, /) -> bytes:
return struct.pack(self._fmt8s, x)
return self._fmt8s.pack(x)
def unpack8(self, x: ReadableBuffer, /) -> int:
return struct.unpack(self._fmt8, x)[0]
return self._fmt8.unpack(x)[0]
def unpack8s(self, x: ReadableBuffer, /) -> int:
return struct.unpack(self._fmt8s, x)[0]
return self._fmt8s.unpack(x)[0]
+3 -1
View File
@@ -183,6 +183,7 @@ class QlGdb(QlDebugger):
from unicorn.arm_const import UC_ARM_REG_R11
from unicorn.arm64_const import UC_ARM64_REG_X29
from unicorn.mips_const import UC_MIPS_REG_INVALID
from unicorn.ppc_const import UC_PPC_REG_31
arch_uc_bp = {
QL_ARCH.X86 : UC_X86_REG_EBP,
@@ -191,7 +192,8 @@ class QlGdb(QlDebugger):
QL_ARCH.ARM64 : UC_ARM64_REG_X29,
QL_ARCH.MIPS : UC_MIPS_REG_INVALID, # skipped
QL_ARCH.A8086 : UC_X86_REG_EBP,
QL_ARCH.CORTEX_M : UC_ARM_REG_R11
QL_ARCH.CORTEX_M : UC_ARM_REG_R11,
QL_ARCH.PPC : UC_PPC_REG_31
}[self.ql.arch.type]
def __get_reg_idx(ucreg: int) -> int:
@@ -25,4 +25,10 @@
<reg name="pc" bitsize="32" type="code_ptr"/>
<reg name="xpsr" bitsize="32" regnum="25"/>
</feature>
<reg name="msp" bitsize="32"/>
<reg name="psp" bitsize="32"/>
<reg name="primask" bitsize="32"/>
<reg name="basepri" bitsize="32"/>
<reg name="faultmask" bitsize="32"/>
<reg name="control" bitsize="32"/>
</feature>
@@ -0,0 +1,12 @@
<?xml version="1.0"?>
<!-- Copyright (C) 2009-2016 Free Software Foundation, Inc.
*!Copying and distribution of this file, with or without modification,
*!are permitted in any medium without royalty provided the copyright
*!notice and this notice are preserved. -->
<!DOCTYPE target SYSTEM "gdb-target.dtd">
<target xmlns:xi="http://www.w3.org/2001/XInclude">
<architecture>armv7-m</architecture>
<xi:include href="arm-m-profile.xml"/>
</target>
+51
View File
@@ -0,0 +1,51 @@
<?xml version="1.0"?>
<!-- Copyright (C) 2007-2020 Free Software Foundation, Inc.
Copying and distribution of this file, with or without modification,
are permitted in any medium without royalty provided the copyright
notice and this notice are preserved. -->
<!DOCTYPE feature SYSTEM "gdb-target.dtd">
<feature name="org.gnu.gdb.power.core">
<reg name="r0" bitsize="32" type="uint32"/>
<reg name="r1" bitsize="32" type="uint32"/>
<reg name="r2" bitsize="32" type="uint32"/>
<reg name="r3" bitsize="32" type="uint32"/>
<reg name="r4" bitsize="32" type="uint32"/>
<reg name="r5" bitsize="32" type="uint32"/>
<reg name="r6" bitsize="32" type="uint32"/>
<reg name="r7" bitsize="32" type="uint32"/>
<reg name="r8" bitsize="32" type="uint32"/>
<reg name="r9" bitsize="32" type="uint32"/>
<reg name="r10" bitsize="32" type="uint32"/>
<reg name="r11" bitsize="32" type="uint32"/>
<reg name="r12" bitsize="32" type="uint32"/>
<reg name="r13" bitsize="32" type="uint32"/>
<reg name="r14" bitsize="32" type="uint32"/>
<reg name="r15" bitsize="32" type="uint32"/>
<reg name="r16" bitsize="32" type="uint32"/>
<reg name="r17" bitsize="32" type="uint32"/>
<reg name="r18" bitsize="32" type="uint32"/>
<reg name="r19" bitsize="32" type="uint32"/>
<reg name="r20" bitsize="32" type="uint32"/>
<reg name="r21" bitsize="32" type="uint32"/>
<reg name="r22" bitsize="32" type="uint32"/>
<reg name="r23" bitsize="32" type="uint32"/>
<reg name="r24" bitsize="32" type="uint32"/>
<reg name="r25" bitsize="32" type="uint32"/>
<reg name="r26" bitsize="32" type="uint32"/>
<reg name="r27" bitsize="32" type="uint32"/>
<reg name="r28" bitsize="32" type="uint32"/>
<reg name="r29" bitsize="32" type="uint32"/>
<reg name="r30" bitsize="32" type="uint32"/>
<reg name="r31" bitsize="32" type="uint32"/>
<reg name="cr" bitsize="32" type="uint32"/>
<reg name="lr" bitsize="32" type="code_ptr"/>
<reg name="pc" bitsize="32" type="code_ptr"/>
<reg name="msr" bitsize="32" type="uint32"/>
<reg name="ctr" bitsize="32" type="uint32"/>
<reg name="xer" bitsize="32" type="uint32"/>
</feature>
+12
View File
@@ -0,0 +1,12 @@
<?xml version="1.0"?>
<!-- Copyright (C) 2009-2016 Free Software Foundation, Inc.
*!Copying and distribution of this file, with or without modification,
*!are permitted in any medium without royalty provided the copyright
*!notice and this notice are preserved. -->
<!DOCTYPE target SYSTEM "gdb-target.dtd">
<target xmlns:xi="http://www.w3.org/2001/XInclude">
<architecture>powerpc:common</architecture>
<xi:include href="ppc-core.xml"/>
</target>
+14 -2
View File
@@ -13,14 +13,21 @@ from qiling.arch.arm_const import (
reg_map_q as arm_regs_q,
reg_map_s as arm_regs_s
)
from qiling.arch.cortex_m_const import (
reg_map as cortex_m_regs
)
from qiling.arch.arm64_const import (
reg_map as arm64_regs,
reg_map_v as arm64_regs_v,
reg_map_fp as arm64_reg_map_fp
)
from qiling.arch.mips_const import (
reg_map as mips_regs_gpr
)
from qiling.arch.x86_const import (
reg_map_32 as x86_regs_32,
reg_map_64 as x86_regs_64,
@@ -31,6 +38,10 @@ from qiling.arch.x86_const import (
reg_map_ymm as x86_regs_ymm
)
from qiling.arch.ppc_const import (
reg_map as ppc_regs
)
from qiling.const import QL_ARCH, QL_OS
RegEntry = Tuple[Optional[int], int, int]
@@ -133,9 +144,10 @@ class QlGdbFeatures:
QL_ARCH.X86: dict(**x86_regs_32, **x86_regs_misc, **x86_regs_cr, **x86_regs_st, **x86_regs_xmm),
QL_ARCH.X8664: dict(**x86_regs_64, **x86_regs_misc, **x86_regs_cr, **x86_regs_st, **x86_regs_xmm, **x86_regs_ymm),
QL_ARCH.ARM: dict(**arm_regs, **arm_regs_vfp, **arm_regs_q, **arm_regs_s),
QL_ARCH.CORTEX_M: arm_regs,
QL_ARCH.CORTEX_M: dict(**cortex_m_regs),
QL_ARCH.ARM64: dict(**arm64_regs, **arm64_regs_v, **arm64_reg_map_fp),
QL_ARCH.MIPS: dict(**mips_regs_gpr)
QL_ARCH.MIPS: dict(**mips_regs_gpr),
QL_ARCH.PPC: dict(**ppc_regs)
}[archtype]
regsinfo = sorted(QlGdbFeatures.__walk_xml_regs(xmltree))
+5 -1
View File
@@ -3,12 +3,14 @@
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
from typing import Dict, Optional
from typing import ClassVar, Dict, Optional
from .arch import Arch
class ArchARM(Arch):
_flags_reg: ClassVar[str] = 'cpsr'
def __init__(self) -> None:
regs = (
'r0', 'r1', 'r2', 'r3',
@@ -134,6 +136,8 @@ class ArchARM(Arch):
class ArchCORTEX_M(ArchARM):
_flags_reg: ClassVar[str] = 'xpsr'
def __init__(self):
super().__init__()
@@ -17,7 +17,7 @@ from capstone.arm_const import (
from unicorn.arm_const import UC_ARM_REG_PC
from .branch_predictor import BranchPredictor, Prophecy
from ..arch import ArchARM
from ..arch import ArchARM, ArchCORTEX_M
from ..misc import InvalidInsn
@@ -27,11 +27,11 @@ class BranchPredictorARM(BranchPredictor, ArchARM):
stop = 'udf'
def get_cpsr(self) -> Tuple[bool, bool, bool, bool]:
"""Get flags map of CPSR.
def get_cond_flags(self) -> Tuple[bool, bool, bool, bool]:
"""Get condition status flags from CPSR / xPSR.
"""
cpsr = self.read_reg('cpsr')
cpsr = self.read_reg(self._flags_reg)
return (
(cpsr & (0b1 << 28)) != 0, # V, overflow flag
@@ -122,9 +122,9 @@ class BranchPredictorARM(BranchPredictor, ArchARM):
def __is_taken(cc: int) -> Tuple[bool, Tuple[bool, ...]]:
pred = predicate[cc]
cpsr = self.get_cpsr()
flags = self.get_cond_flags()
return pred(*cpsr), cpsr
return pred(*flags), flags
# conditions predicate selector
predicate: Dict[int, Callable[..., bool]] = {
@@ -215,13 +215,13 @@ class BranchPredictorARM(BranchPredictor, ArchARM):
where = __parse_op(operands[1], **msize[suffix])
elif iname in binop:
going, cpsr = __is_taken(insn.cc)
going, flags = __is_taken(insn.cc)
if going:
operator = binop[iname]
op1 = __parse_op(operands[1])
op2 = __parse_op(operands[2])
carry = int(cpsr[1])
carry = int(flags[1])
where = (op1 and op2) and operator(op1, op2, carry)
@@ -261,6 +261,6 @@ class BranchPredictorARM(BranchPredictor, ArchARM):
return Prophecy(going, where)
class BranchPredictorCORTEX_M(BranchPredictorARM):
class BranchPredictorCORTEX_M(BranchPredictorARM, ArchCORTEX_M):
"""Branch Predictor for ARM Cortex-M.
"""
+2 -2
View File
@@ -57,7 +57,7 @@ class Context:
"""Helper function for disassembling.
"""
insn_bytes = self.read_insn(address)
insn_bytes = self.read_insn(address) or b''
insn = None
if insn_bytes:
@@ -75,7 +75,7 @@ class Context:
A tuple of: instruction address, size, mnemonic and operands
"""
insn_bytes = self.read_insn(address)
insn_bytes = self.read_insn(address) or b''
insn = None
if insn_bytes:
+1 -4
View File
@@ -136,10 +136,7 @@ class QlQdb(Cmd, QlDebugger):
with self.__set_temp(self.ql, 'verbose', QL_VERBOSE.DISABLED):
self.ql.os.run()
if self.ql.os.type is QL_OS.BLOB:
self.ql.loader.entry_point = self.ql.loader.load_address
elif init_hook:
if init_hook:
for each_hook in init_hook:
self.do_breakpoint(each_hook)
+2 -2
View File
@@ -3,7 +3,7 @@
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
from typing import Iterator, Optional
from typing import Iterator
from .render import Render, ContextRender
from ..arch import ArchARM, ArchCORTEX_M
@@ -15,7 +15,7 @@ class ContextRenderARM(ContextRender, ArchARM):
"""
def print_mode_info(self) -> None:
cpsr = self.read_reg('cpsr')
cpsr = self.read_reg(self._flags_reg)
flags = ArchARM.get_flags(cpsr)
mode = ArchARM.get_mode(cpsr)
+2 -2
View File
@@ -96,8 +96,8 @@ def ql_afl_fuzz_custom(ql: Qiling,
def __place_input_wrapper(uc: Uc, input_bytes: Array[c_char], iters: int, context: Any) -> bool:
return place_input_callback(ql, input_bytes.raw, iters)
def __validate_crash_wrapper(uc: Uc, result: int, input_bytes: bytes, iters: int, context: Any) -> bool:
return validate_crash_callback(ql, result, input_bytes, iters)
def __validate_crash_wrapper(uc: Uc, result: int, input_bytes: Array[c_char], iters: int, context: Any) -> bool:
return validate_crash_callback(ql, result, input_bytes.raw, iters)
def __fuzzing_wrapper(uc: Uc, context: Any) -> int:
return fuzzing_callback(ql)
+12 -11
View File
@@ -3,9 +3,14 @@
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
from abc import ABC, abstractmethod
from __future__ import annotations
from qiling import Qiling
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from qiling import Qiling
class QlBaseCoverage(ABC):
@@ -15,25 +20,21 @@ class QlBaseCoverage(ABC):
all the methods marked with the @abstractmethod decorator.
"""
FORMAT_NAME: str
def __init__(self, ql: Qiling):
super().__init__()
self.ql = ql
@property
@staticmethod
@abstractmethod
def FORMAT_NAME() -> str:
raise NotImplementedError
@abstractmethod
def activate(self):
def activate(self) -> None:
pass
@abstractmethod
def deactivate(self):
def deactivate(self) -> None:
pass
@abstractmethod
def dump_coverage(self, coverage_file: str):
def dump_coverage(self, coverage_file: str) -> None:
pass
+56 -23
View File
@@ -3,12 +3,20 @@
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
from ctypes import Structure
from ctypes import c_uint32, c_uint16
from __future__ import annotations
from ctypes import Structure, c_uint32, c_uint16
from functools import lru_cache
from typing import TYPE_CHECKING, BinaryIO, Dict, Tuple
from .base import QlBaseCoverage
if TYPE_CHECKING:
from qiling import Qiling
from qiling.loader.loader import QlLoader
# Adapted from https://www.ayrx.me/drcov-file-format
class bb_entry(Structure):
_fields_ = [
@@ -29,36 +37,61 @@ class QlDrCoverage(QlBaseCoverage):
FORMAT_NAME = "drcov"
def __init__(self, ql):
def __init__(self, ql: Qiling):
super().__init__(ql)
self.drcov_version = 2
self.drcov_flavor = 'drcov'
self.basic_blocks = []
self.basic_blocks: Dict[int, bb_entry] = {}
self.bb_callback = None
@staticmethod
def block_callback(ql, address, size, self):
for mod_id, mod in enumerate(ql.loader.images):
if mod.base <= address <= mod.end:
ent = bb_entry(address - mod.base, size, mod_id)
self.basic_blocks.append(ent)
break
@lru_cache(maxsize=64)
def _get_img_base(self, loader: QlLoader, address: int) -> Tuple[int, int]:
"""Retrieve the containing image of a given address.
def activate(self):
self.bb_callback = self.ql.hook_block(self.block_callback, user_data=self)
Addresses are expected to be aligned to page boundary, and cached for faster retrieval.
"""
def deactivate(self):
self.ql.hook_del(self.bb_callback)
return next((i, img.base) for i, img in enumerate(loader.images) if img.base <= address < img.end)
def block_callback(self, ql: Qiling, address: int, size: int):
if address not in self.basic_blocks:
try:
# we rely on the fact that images are allocated on page size boundary and
# use it to speed up image retrieval. we align the basic block address to
# page boundary, knowing basic blocks within the same page belong to the
# same image. then we use the aligned address to retreive the containing
# image. returned values are cached so subsequent retrievals for basic
# blocks within the same page will return the cached value instead of
# going through the retreival process again (up to maxsize cached pages)
i, img_base = self._get_img_base(ql.loader, address & ~(0x1000 - 1))
except StopIteration:
pass
else:
self.basic_blocks[address] = bb_entry(address - img_base, size, i)
def activate(self) -> None:
self.bb_callback = self.ql.hook_block(self.block_callback)
def deactivate(self) -> None:
if self.bb_callback:
self.ql.hook_del(self.bb_callback)
def dump_coverage(self, coverage_file: str) -> None:
def __write_line(bio: BinaryIO, line: str) -> None:
bio.write(f'{line}\n'.encode())
def dump_coverage(self, coverage_file):
with open(coverage_file, "wb") as cov:
cov.write(f"DRCOV VERSION: {self.drcov_version}\n".encode())
cov.write(f"DRCOV FLAVOR: {self.drcov_flavor}\n".encode())
cov.write(f"Module Table: version {self.drcov_version}, count {len(self.ql.loader.images)}\n".encode())
cov.write("Columns: id, base, end, entry, checksum, timestamp, path\n".encode())
__write_line(cov, f"DRCOV VERSION: {self.drcov_version}")
__write_line(cov, f"DRCOV FLAVOR: {self.drcov_flavor}")
__write_line(cov, f"Module Table: version {self.drcov_version}, count {len(self.ql.loader.images)}")
__write_line(cov, "Columns: id, base, end, entry, checksum, timestamp, path")
for mod_id, mod in enumerate(self. ql.loader.images):
cov.write(f"{mod_id}, {mod.base}, {mod.end}, 0, 0, 0, {mod.path}\n".encode())
cov.write(f"BB Table: {len(self.basic_blocks)} bbs\n".encode())
for bb in self.basic_blocks:
__write_line(cov, f"{mod_id}, {mod.base}, {mod.end}, 0, 0, 0, {mod.path}")
__write_line(cov, f"BB Table: {len(self.basic_blocks)} bbs")
for bb in self.basic_blocks.values():
cov.write(bytes(bb))
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
#
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
@@ -17,10 +17,6 @@ class QlDrCoverageExact(QlDrCoverage):
FORMAT_NAME = "drcov_exact"
def __init__(self, ql):
super().__init__(ql)
def activate(self):
def activate(self) -> None:
# We treat every instruction as a block on its own.
self.bb_callback = self.ql.hook_code(self.block_callback, user_data=self)
self.bb_callback = self.ql.hook_code(self.block_callback)
+35 -23
View File
@@ -1,19 +1,30 @@
#!/usr/bin/env python3
#
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
from collections import namedtuple
from os.path import basename
from __future__ import annotations
import os
from typing import Any, TYPE_CHECKING, List, NamedTuple
from .base import QlBaseCoverage
if TYPE_CHECKING:
from qiling import Qiling
# Adapted from https://github.com/nccgroup/Cartographer/blob/main/EZCOV.md#coverage-data
class bb_entry(namedtuple('bb_entry', 'offset size mod_id')):
def csvline(self):
offset = '0x{:08x}'.format(self.offset)
class bb_entry(NamedTuple):
offset: int
size: int
mod_id: Any
def as_csv(self) -> str:
offset = f'{self.offset:#010x}'
mod_id = f"[ {self.mod_id if self.mod_id is not None else ''} ]"
return f"{offset},{self.size},{mod_id}\n"
class QlEzCoverage(QlBaseCoverage):
@@ -27,29 +38,30 @@ class QlEzCoverage(QlBaseCoverage):
FORMAT_NAME = "ezcov"
def __init__(self, ql):
def __init__(self, ql: Qiling):
super().__init__(ql)
self.ezcov_version = 1
self.ezcov_flavor = 'ezcov'
self.basic_blocks = []
self.bb_callback = None
self.ezcov_flavor = 'ezcov'
self.basic_blocks: List[bb_entry] = []
self.bb_callback = None
@staticmethod
def block_callback(ql, address, size, self):
mod = ql.loader.find_containing_image(address)
if mod is not None:
ent = bb_entry(address - mod.base, size, basename(mod.path))
self.basic_blocks.append(ent)
def block_callback(self, ql: Qiling, address: int, size: int):
img = ql.loader.find_containing_image(address)
def activate(self):
self.bb_callback = self.ql.hook_block(self.block_callback, user_data=self)
if img is not None:
self.basic_blocks.append(bb_entry(address - img.base, size, os.path.basename(img.path)))
def deactivate(self):
self.ql.hook_del(self.bb_callback)
def activate(self) -> None:
self.bb_callback = self.ql.hook_block(self.block_callback)
def dump_coverage(self, coverage_file):
def deactivate(self) -> None:
if self.bb_callback:
self.ql.hook_del(self.bb_callback)
def dump_coverage(self, coverage_file: str) -> None:
with open(coverage_file, "w") as cov:
cov.write(f"EZCOV VERSION: {self.ezcov_version}\n")
cov.write("# Qiling EZCOV exporter tool\n")
for bb in self.basic_blocks:
cov.write(bb.csvline())
cov.writelines(bb.as_csv() for bb in self.basic_blocks)
+163 -104
View File
@@ -5,7 +5,6 @@ UseAsScript = True
import sys
import collections
import time
import struct
import re
import logging
@@ -37,9 +36,6 @@ import ida_auto
import ida_netnode
import ida_hexrays
import ida_range
# PyQt
from PyQt5 import QtCore, QtWidgets
from PyQt5.QtWidgets import (QPushButton, QHBoxLayout)
# Qiling
from qiling import Qiling
@@ -55,7 +51,6 @@ from qiling import __version__ as QLVERSION
from qiling.os.filestruct import ql_file
from keystone import *
QilingHomePage = 'https://www.qiling.io'
QilingStableVersionURL = 'https://raw.githubusercontent.com/qilingframework/qiling/master/qiling/__version__.py'
logging.basicConfig(level=logging.INFO, format='[%(levelname)s][%(module)s:%(lineno)d] %(message)s')
@@ -69,7 +64,27 @@ class Colors(Enum):
Gray = 0xd9d9d9
Beige = 0xCCF2FF
class IDA:
def _load_qt_bindings():
if IDA_SDK_VERSION >= 900:
try:
from PySide6 import QtCore, QtWidgets
from PySide6.QtWidgets import (QPushButton, QHBoxLayout)
logging.info("Using PySide6 for Qt bindings (IDA >= 9).")
return QtCore, QtWidgets, QPushButton, QHBoxLayout
except Exception as e:
logging.warning("Failed to import PySide6: %s. Trying PyQt5 fallback.", e)
try:
from PyQt5 import QtCore, QtWidgets
from PyQt5.QtWidgets import (QPushButton, QHBoxLayout)
logging.info("Using PyQt5 for Qt bindings (IDA < 9 or fallback).")
return QtCore, QtWidgets, QPushButton, QHBoxLayout
except Exception as e:
logging.error("Failed to import PyQt bindings: %s", e)
raise
QtCore, QtWidgets, QPushButton, QHBoxLayout = _load_qt_bindings()
class IDABase:
def __init__(self):
pass
@@ -79,15 +94,15 @@ class IDA:
@staticmethod
def get_function_start(addr):
return IDA.get_function(addr).start_ea
return IDABase.get_function(addr).start_ea
@staticmethod
def get_function_end(addr):
return IDA.get_function(addr).end_ea
return IDABase.get_function(addr).end_ea
@staticmethod
def get_function_framesize(addr):
return IDA.get_function(addr).frsize
return IDABase.get_function(addr).frsize
@staticmethod
def get_function_name(addr):
@@ -95,7 +110,7 @@ class IDA:
@staticmethod
def get_functions():
return [IDA.get_function(func) for func in idautils.Functions()]
return [IDABase.get_function(func) for func in idautils.Functions()]
@staticmethod
def set_color(addr, what, color):
@@ -104,7 +119,7 @@ class IDA:
@staticmethod
def color_block(bb, color):
for i in range(bb.start_ea, bb.end_ea):
IDA.set_color(i, idc.CIC_ITEM, color)
IDABase.set_color(i, idc.CIC_ITEM, color)
# note:
# corresponds to IDA graph view
@@ -113,8 +128,8 @@ class IDA:
# arg can be a function or a (start, end) tuple or an address in the function
@staticmethod
def get_flowchart(arg):
if type(arg) is int:
func = IDA.get_function(arg)
if isinstance(arg, int):
func = IDABase.get_function(arg)
if func is None:
return None
return ida_gdl.FlowChart(func)
@@ -122,7 +137,9 @@ class IDA:
@staticmethod
def get_block(addr):
flowchart = IDA.get_flowchart(addr)
flowchart = IDABase.get_flowchart(addr)
if flowchart is None:
return None
for bb in flowchart:
if bb.start_ea <= addr and addr < bb.end_ea:
return bb
@@ -143,10 +160,10 @@ class IDA:
@staticmethod
def get_starting_block(addr):
flowchart = IDA.get_flowchart(addr)
flowchart = IDABase.get_flowchart(addr)
if flowchart is None:
return None
func = IDA.get_function(addr)
func = IDABase.get_function(addr)
for bb in flowchart:
if bb.start_ea == func.start_ea:
return bb
@@ -154,8 +171,10 @@ class IDA:
@staticmethod
def get_terminating_blocks(addr):
flowchart = IDA.get_flowchart(addr)
return [bb for bb in flowchart if IDA.block_is_terminating(bb)]
flowchart = IDABase.get_flowchart(addr)
if flowchart is None:
return []
return [bb for bb in flowchart if IDABase.block_is_terminating(bb)]
@staticmethod
def get_prev_head(addr, minea=0):
@@ -180,46 +199,45 @@ class IDA:
@staticmethod
def __addr_in_seg(addr):
segs = IDA.get_segments()
segs = IDABase.get_segments()
for seg in segs:
if addr < seg.end_ea and addr >= seg.start_ea:
return seg
return None
# note: accept name and address in the segment
@staticmethod
def get_segment(arg):
if type(arg) is int:
return IDA.__addr_in_seg(arg)
else: # str
return IDA.get_segment_by_name(arg)
if isinstance(arg, int):
return IDABase.__addr_in_seg(arg)
else:
return IDABase.get_segment_by_name(arg)
@staticmethod
def get_segment_start(arg):
seg = IDA.get_segment(arg)
seg = IDABase.get_segment(arg)
if seg is not None:
return seg.start_ea
return None
@staticmethod
def get_segment_end(arg):
seg = IDA.get_segment(arg)
seg = IDABase.get_segment(arg)
if seg is not None:
return seg.end_ea
return None
@staticmethod
def get_segment_perm(arg):
seg = IDA.get_segment(arg)
seg = IDABase.get_segment(arg)
if seg is not None:
return seg.perm # RWX e.g. 0b101 = R + X
return seg.perm
return None
@staticmethod
def get_segment_type(arg):
seg = IDA.get_segment(arg)
seg = IDABase.get_segment(arg)
if seg is not None:
return seg.type # 0x1 SEG_DATA 0x2 SEG_CODE See doc for details
return seg.type
return None
@staticmethod
@@ -229,12 +247,10 @@ class IDA:
return None
return r
# immidiate value
@staticmethod
def get_operand(addr, n):
return (idc.get_operand_type(addr, n), idc.get_operand_value(addr, n))
# eax, ecx, etc
@staticmethod
def print_operand(addr, n):
return idc.print_operand(addr, n)
@@ -248,7 +264,7 @@ class IDA:
p = begin
cnt = 0
while p < end:
sz = IDA.get_instruction_size(p)
sz = IDABase.get_instruction_size(p)
cnt += 1
p += sz
return cnt
@@ -293,96 +309,34 @@ class IDA:
def get_input_file_path():
return ida_nalt.get_input_file_path()
@staticmethod
def get_info_structure():
return ida_idaapi.get_inf_structure()
@staticmethod
def get_main_address():
return IDA.get_info_structure().main
@staticmethod
def get_max_address():
return IDA.get_info_structure().max_ea
@staticmethod
def get_min_address():
return IDA.get_info_structure().min_ea
@staticmethod
def is_big_endian():
return IDA.get_info_structure().is_be()
@staticmethod
def is_little_endian():
return not IDA.is_big_endian()
@staticmethod
def get_filetype():
info = IDA.get_info_structure()
ftype = info.filetype
if ftype == ida_ida.f_MACHO:
return "macho"
elif ftype == ida_ida.f_PE or ftype == ida_ida.f_EXE or ftype == ida_ida.f_EXE_old: # is this correct?
return "pe"
elif ftype == ida_ida.f_ELF:
return "elf"
else:
return None
@staticmethod
def get_ql_arch_string():
info = IDA.get_info_structure()
proc = info.procname.lower()
result = None
if proc == "metapc":
result = "x86"
if info.is_64bit():
result = "x8664"
elif "mips" in proc:
result = "mips"
elif "arm" in proc:
result = "arm32"
if info.is_64bit():
result = "arm64"
# That's all we support :(
return result
@staticmethod
def get_current_address():
return ida_kernwin.get_screen_ea()
# return (?, start, end)
@staticmethod
def get_last_selection():
return ida_kernwin.read_range_selection(None)
# Use with skipcalls
# note that the address is the end of target instruction
# e.g.:
# 0x1 push eax
# 0x4 mov eax, 0
# call get_frame_sp_delta(0x4) and get -4.
@staticmethod
def get_frame_sp_delta(addr):
return ida_frame.get_sp_delta(IDA.get_function(addr), addr)
return ida_frame.get_sp_delta(IDABase.get_function(addr), addr)
@staticmethod
def patch_bytes(addr, bs):
return ida_bytes.patch_bytes(addr, bs)
@staticmethod
def fill_bytes(start, end, bs = b'\x90'):
def fill_bytes(start, end, bs=b'\x90'):
return ida_bytes.patch_bytes(start, bs*(end-start))
@staticmethod
def nop_selection():
_, start, end = IDA.get_last_selection()
return IDA.fill_bytes(start, end)
_, start, end = IDABase.get_last_selection()
return IDABase.fill_bytes(start, end)
@staticmethod
def fill_block(bb, bs=b'\x90'):
return IDA.fill_bytes(bb.start_ea, bb.end_ea, bs)
return IDABase.fill_bytes(bb.start_ea, bb.end_ea, bs)
@staticmethod
def assemble(ea, cs, ip, use32, line):
@@ -394,7 +348,7 @@ class IDA:
@staticmethod
def create_bytes_array(start, end):
return IDA.create_data(start, ida_bytes.byte_flag(), end-start)
return IDABase.create_data(start, ida_bytes.byte_flag(), end-start)
@staticmethod
def create_byte(ea, length, force=False):
@@ -418,13 +372,12 @@ class IDA:
@staticmethod
def get_item(ea):
return (IDA.get_item_head(ea), IDA.get_item_end(ea))
return (IDABase.get_item_head(ea), IDABase.get_item_end(ea))
@staticmethod
def is_colored_item(ea):
return ida_nalt.is_colored_item(ea)
# NOTE: The [start, end) range should include all control flows except long calls.
@staticmethod
def get_micro_code_mba(start, end, decomp_flags=ida_hexrays.DECOMP_WARNINGS, maturity=7):
mbrgs = ida_hexrays.mba_ranges_t()
@@ -444,6 +397,112 @@ class IDA:
cur = cur.next
return
class IDA7(IDABase):
@staticmethod
def get_info_structure():
return ida_idaapi.get_inf_structure()
@staticmethod
def get_main_address():
return IDA7.get_info_structure().main
@staticmethod
def get_max_address():
return IDA7.get_info_structure().max_ea
@staticmethod
def get_min_address():
return IDA7.get_info_structure().min_ea
@staticmethod
def is_big_endian():
return IDA7.get_info_structure().is_be
@staticmethod
def is_little_endian():
return not IDA7.is_big_endian()
@staticmethod
def get_filetype():
ftype = IDA7.get_info_structure().filetype
if ftype in (ida_ida.f_PE, ida_ida.f_EXE, ida_ida.f_EXE_old):
return "pe"
elif ftype == ida_ida.f_MACHO:
return "macho"
elif ftype == ida_ida.f_ELF:
return "elf"
return None
@staticmethod
def get_ql_arch_string():
proc = IDA7.get_info_structure().procname.lower()
is_64_bit = IDA7.get_info_structure().is_64bit()
if proc == "metapc":
return "x8664" if is_64_bit else "x86"
if "mips" in proc:
return "mips"
if "arm" in proc:
return "arm64" if is_64_bit else "arm32"
return None
class IDA9(IDABase):
@staticmethod
def get_info_structure():
return ida_idaapi.get_inf_structure()
@staticmethod
def get_main_address():
return ida_ida.inf_get_main()
@staticmethod
def get_max_address():
return ida_ida.inf_get_max_ea()
@staticmethod
def get_min_address():
return ida_ida.inf_get_min_ea()
@staticmethod
def is_big_endian():
return ida_ida.inf_is_be()
@staticmethod
def is_little_endian():
return not ida_ida.inf_is_be()
@staticmethod
def get_filetype():
ftype = ida_ida.inf_get_filetype()
if ftype in (ida_ida.f_PE, ida_ida.f_EXE, ida_ida.f_EXE_old):
return "pe"
elif ftype == ida_ida.f_MACHO:
return "macho"
elif ftype == ida_ida.f_ELF:
return "elf"
return None
@staticmethod
def get_ql_arch_string():
proc = ida_ida.inf_get_procname().lower()
is_64_bit = ida_ida.inf_is_64bit()
if proc == "metapc":
return "x8664" if is_64_bit else "x86"
if "mips" in proc:
return "mips"
if "arm" in proc:
return "arm64" if is_64_bit else "arm32"
return None
def get_ida_instance():
if IDA_SDK_VERSION >= 900:
logging.info("Using IDA9 compatibility layer")
return IDA9()
else:
logging.info("Using IDA7 compatibility layer")
return IDA7()
IDA = get_ida_instance()
### View Class
class QlEmuRegView(simplecustviewer_t):
@@ -1006,7 +1065,7 @@ class QlEmuPlugin(plugin_t, UI_Hooks):
def init(self):
# init data
logging.info('---------------------------------------------------------------------------------------')
logging.info('Qiling Emulator Plugin For IDA, by Qiling Team. Version {0}, 2020'.format(QLVERSION))
logging.info('Qiling Emulator Plugin For IDA, by Qiling Team. Version {0}, 2025'.format(QLVERSION))
logging.info('Based on Qiling v{0}'.format(QLVERSION))
logging.info('Find more information about Qiling at https://qiling.io')
logging.info('---------------------------------------------------------------------------------------')
+7 -11
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
#
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
# This code structure is copied and modified from the coverage extension
@@ -12,24 +12,20 @@ class QlBaseTrace(ABC):
To add support for a new coverage format, just derive from this class and implement
all the methods marked with the @abstractmethod decorator.
"""
FORMAT_NAME: str
def __init__(self):
super().__init__()
@property
@staticmethod
@abstractmethod
def FORMAT_NAME():
raise NotImplementedError
@abstractmethod
def activate(self):
def activate(self) -> None:
pass
@abstractmethod
def deactivate(self):
def deactivate(self) -> None:
pass
@abstractmethod
def dump_trace(self, trace_file):
def dump_trace(self, trace_file: str) -> None:
pass
+108 -59
View File
@@ -1,26 +1,70 @@
#!/usr/bin/env python3
#
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
import ctypes
from functools import cached_property
from typing import Any, Dict, List, Optional, Tuple
from qiling.core import Qiling
from qiling import Qiling
from qiling.hw.peripheral import QlPeripheral
from qiling.utils import ql_get_module_function
from qiling.exception import QlErrorModuleFunctionNotFound
# should adhere to the QlMmioHandler interface, but not extend it directly to
# avoid potential pickling issues
class QlPripheralHandler:
def __init__(self, hwman: "QlHwManager", base: int, size: int, label: str) -> None:
self._hwman = hwman
self._base = base
self._size = size
self._label = label
def __getstate__(self):
state = self.__dict__.copy()
del state['_hwman'] # remove non-pickleable reference
return state
@cached_property
def _mmio(self) -> bytearray:
"""Get memory buffer used to back non-mapped hardware mmio regions.
"""
return bytearray(self._size)
def read(self, ql: Qiling, offset: int, size: int) -> int:
address = self._base + offset
hardware = self._hwman.find(address)
if hardware:
return hardware.read(address - hardware.base, size)
else:
ql.log.debug('[%s] read non-mapped hardware [%#010x]', self._label, address)
return int.from_bytes(self._mmio[offset:offset + size], byteorder='little')
def write(self, ql: Qiling, offset: int, size: int, value: int) -> None:
address = self._base + offset
hardware = self._hwman.find(address)
if hardware:
hardware.write(address - hardware.base, size, value)
else:
ql.log.debug('[%s] write non-mapped hardware [%#010x] = %#010x', self._label, address, value)
self._mmio[offset:offset + size] = value.to_bytes(size, 'little')
class QlHwManager:
def __init__(self, ql: Qiling):
self.ql = ql
self.entity = {}
self.region = {}
self.entity: Dict[str, QlPeripheral] = {}
self.region: Dict[str, List[Tuple[int, int]]] = {}
self.stepable = {}
def create(self, label: str, struct: str=None, base: int=None, kwargs: dict={}) -> "QlPeripheral":
def create(self, label: str, struct: Optional[str] = None, base: Optional[int] = None, kwargs: Optional[Dict[str, Any]] = None) -> QlPeripheral:
""" Create the peripheral accroding the label and envs.
struct: Structure of the peripheral. Use defualt ql structure if not provide.
@@ -30,39 +74,45 @@ class QlHwManager:
if struct is None:
struct, base, kwargs = self.load_env(label.upper())
try:
entity = ql_get_module_function('qiling.hw', struct)(self.ql, label, **kwargs)
self.entity[label] = entity
if hasattr(entity, 'step'):
self.stepable[label] = entity
if kwargs is None:
kwargs = {}
try:
entity = ql_get_module_function('qiling.hw', struct)(self.ql, label, **kwargs)
except QlErrorModuleFunctionNotFound:
self.ql.log.warning(f'could not create {struct}({label}): implementation not found')
else:
assert isinstance(entity, QlPeripheral)
assert isinstance(base, int)
self.entity[label] = entity
self.region[label] = [(lbound + base, rbound + base) for (lbound, rbound) in entity.region]
return entity
except QlErrorModuleFunctionNotFound:
self.ql.log.debug(f'The {struct}({label}) has not been implemented')
def delete(self, label: str):
# FIXME: what should we do if struct is not implemented? is it OK to return None , or we fail?
def delete(self, label: str) -> None:
""" Remove the peripheral
"""
if label in self.entity:
self.entity.pop(label)
self.region.pop(label)
if label in self.stepable:
self.stepable.pop(label)
def load_env(self, label: str):
if label in self.entity:
del self.entity[label]
if label in self.region:
del self.region[label]
def load_env(self, label: str) -> Tuple[str, int, Dict[str, Any]]:
""" Get peripheral information (structure, base address, initialization list) from env.
Args:
label (str): Peripheral Label
"""
args = self.ql.env[label]
return args['struct'], args['base'], args.get("kwargs", {})
def load_all(self):
@@ -70,48 +120,30 @@ class QlHwManager:
if args['type'] == 'peripheral':
self.create(label.lower(), args['struct'], args['base'], args.get("kwargs", {}))
def find(self, address: int):
# TODO: this is wasteful. device mapping is known at creation time. at least we could cache lru entries
def find(self, address: int) -> Optional[QlPeripheral]:
""" Find the peripheral at `address`
"""
for label in self.entity.keys():
for lbound, rbound in self.region[label]:
if lbound <= address < rbound:
return self.entity[label]
return None
def step(self):
""" Update all peripheral's state
""" Update all peripheral's state
"""
for entity in self.stepable.values():
entity.step()
def setup_mmio(self, begin, size, info=""):
mmio = ctypes.create_string_buffer(size)
for ent in self.entity.values():
if hasattr(ent, 'step'):
ent.step()
def mmio_read_cb(ql, offset, size):
address = begin + offset
hardware = self.find(address)
if hardware:
return hardware.read(address - hardware.base, size)
else:
ql.log.debug('%s Read non-mapped hardware [0x%08x]' % (info, address))
buf = ctypes.create_string_buffer(size)
ctypes.memmove(buf, ctypes.addressof(mmio) + offset, size)
return int.from_bytes(buf.raw, byteorder='little')
def setup_mmio(self, begin: int, size: int, info: str) -> None:
dev = QlPripheralHandler(self, begin, size, info)
def mmio_write_cb(ql, offset, size, value):
address = begin + offset
hardware = self.find(address)
if hardware:
hardware.write(address - hardware.base, size, value)
else:
ql.log.debug('%s Write non-mapped hardware [0x%08x] = 0x%08x' % (info, address, value))
ctypes.memmove(ctypes.addressof(mmio) + offset, (value).to_bytes(size, 'little'), size)
self.ql.mem.map_mmio(begin, size, mmio_read_cb, mmio_write_cb, info=info)
self.ql.mem.map_mmio(begin, size, dev, info)
def show_info(self):
self.ql.log.info(f'{"Start":8s} {"End":8s} {"Label":8s} {"Class"}')
@@ -131,8 +163,25 @@ class QlHwManager:
return self.entity.get(key)
def save(self):
return {label : entity.save() for label, entity in self.entity.items()}
return {
'entity': {label: entity.save() for label, entity in self.entity.items()},
'region': self.region
}
def restore(self, saved_state):
for label, data in saved_state.items():
entity = saved_state['entity']
assert isinstance(entity, dict)
region = saved_state['region']
assert isinstance(region, dict)
for label, data in entity.items():
self.entity[label].restore(data)
self.region = region
# a dirty hack to rehydrate non-pickleable hwman
# a proper fix would require a deeper refactoring to how peripherals are created and managed
for *_, ph in self.ql.mem.map_info:
if isinstance(ph, QlPripheralHandler):
setattr(ph, '_hwman', self)
+13 -10
View File
@@ -4,8 +4,8 @@
#
from qiling import Qiling
from qiling.loader.loader import QlLoader
from qiling.os.memory import QlMemoryHeap
from qiling.loader.loader import QlLoader, Image
class QlLoaderBLOB(QlLoader):
def __init__(self, ql: Qiling):
@@ -14,15 +14,18 @@ class QlLoaderBLOB(QlLoader):
self.load_address = 0
def run(self):
self.load_address = self.ql.os.entry_point # for consistency
self.load_address = self.ql.os.load_address
self.entry_point = self.ql.os.entry_point
self.ql.mem.map(self.ql.os.entry_point, self.ql.os.code_ram_size, info="[code]")
self.ql.mem.write(self.ql.os.entry_point, self.ql.code)
code_begins = self.load_address
code_size = self.ql.os.code_ram_size
code_ends = code_begins + code_size
heap_address = self.ql.os.entry_point + self.ql.os.code_ram_size
heap_size = int(self.ql.os.profile.get("CODE", "heap_size"), 16)
self.ql.os.heap = QlMemoryHeap(self.ql, heap_address, heap_address + heap_size)
self.ql.mem.map(code_begins, code_size, info="[code]")
self.ql.mem.write(code_begins, self.ql.code)
self.ql.arch.regs.arch_sp = heap_address - 0x1000
# allow image-related functionalities
self.images.append(Image(code_begins, code_ends, 'blob_code'))
return
# FIXME: stack pointer should be a configurable profile setting
self.ql.arch.regs.arch_sp = code_ends - 0x1000
+33 -29
View File
@@ -1,7 +1,7 @@
#!/usr/bin/env python3
#
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
# Built on top of Unicorn emulator (www.unicorn-engine.org)
# Built on top of Unicorn emulator (www.unicorn-engine.org)
import io
@@ -27,7 +27,7 @@ class IhexParser:
if addr != begin + len(stream):
self.segments.append((begin, stream))
begin, stream = addr, data
else:
stream += data
@@ -36,13 +36,13 @@ class IhexParser:
def parse_line(self, line):
if len(line) < 9:
return
desc = line[7: 9]
size = int(line[1: 3], 16)
size = int(line[1: 3], 16)
addr = bytes.fromhex(line[3: 7])
data = bytes.fromhex(line[9: 9 + size * 2])
data = bytes.fromhex(line[9: 9 + size * 2])
if desc == '00': # Data
offset = int.from_bytes(addr, byteorder='big')
self.mem.append((self.base + offset, data))
@@ -52,20 +52,20 @@ class IhexParser:
elif desc == '04': # Extended Linear Address
self.base = int.from_bytes(data, byteorder='big') * 0x10000
class QlLoaderMCU(QlLoader):
def __init__(self, ql:Qiling):
super().__init__(ql)
super().__init__(ql)
self.entry_point = 0
self.load_address = 0
self.filetype = self.guess_filetype()
if self.filetype == 'elf':
with open(self.ql.path, 'rb') as infile:
self.elf = ELFFile(io.BytesIO(infile.read()))
elif self.filetype == 'bin':
self.map_address = self.argv[1]
@@ -74,8 +74,8 @@ class QlLoaderMCU(QlLoader):
def guess_filetype(self):
if self.ql.path.endswith('.elf'):
return 'elf'
return 'elf'
if self.ql.path.endswith('.bin'):
return 'bin'
@@ -83,7 +83,7 @@ class QlLoaderMCU(QlLoader):
return 'hex'
return 'elf'
def reset(self):
if self.filetype == 'elf':
for segment in self.elf.iter_segments(type='PT_LOAD'):
@@ -99,7 +99,7 @@ class QlLoaderMCU(QlLoader):
for begin, data in self.ihex.segments:
self.ql.mem.write(begin, data)
self.ql.arch.init_context()
self.entry_point = self.ql.arch.regs.read('pc')
@@ -109,30 +109,34 @@ class QlLoaderMCU(QlLoader):
def load_env(self):
for name, args in self.env.items():
memtype = args['type']
if memtype == 'memory':
size = args['size']
base = args['base']
self.ql.mem.map(base, size, info=f'[{name}]')
if memtype == 'remap':
# elif memtype == 'remap':
# size = args['size']
# base = args['base']
# alias = args['alias']
# self.ql.hw.setup_remap(alias, base, size, info=f'[{name}]')
elif memtype == 'mmio':
size = args['size']
base = args['base']
alias = args['alias']
self.ql.hw.setup_remap(alias, base, size, info=f'[{name}]')
self.ql.hw.setup_mmio(base, size, name)
if memtype == 'mmio':
size = args['size']
base = args['base']
self.ql.hw.setup_mmio(base, size, info=f'[{name}]')
if memtype == 'core':
elif memtype == 'core':
self.ql.hw.create(name.lower())
else:
self.ql.log.debug(f'ignoring unknown memory type "{memtype}" for {name}')
def run(self):
self.load_profile()
self.load_env()
## Handle interrupt from instruction execution
self.ql.hook_intr(self.ql.arch.unicorn_exception_handler)
self.reset()
+1
View File
@@ -644,6 +644,7 @@ class Process:
# DLLs that seem to contain most of the requested symbols
key_dlls = (
'kernel32.dll',
'ntdll.dll',
'kernelbase.dll',
'ucrtbase.dll'
+16 -5
View File
@@ -8,6 +8,8 @@ from qiling.cc import QlCC, intel, arm, mips, riscv, ppc
from qiling.const import QL_ARCH, QL_OS
from qiling.os.fcall import QlFunctionCall
from qiling.os.os import QlOs
from qiling.os.memory import QlMemoryHeap
class QlOsBlob(QlOs):
""" QlOsBlob for bare barines.
@@ -21,7 +23,7 @@ class QlOsBlob(QlOs):
type = QL_OS.BLOB
def __init__(self, ql: Qiling):
super(QlOsBlob, self).__init__(ql)
super().__init__(ql)
self.ql = ql
@@ -39,11 +41,20 @@ class QlOsBlob(QlOs):
self.fcall = QlFunctionCall(ql, cc)
def run(self):
if self.ql.entry_point:
# if entry point was set explicitly, override the default one
if self.ql.entry_point is not None:
self.entry_point = self.ql.entry_point
self.exit_point = self.ql.loader.load_address + len(self.ql.code)
if self.ql.exit_point:
self.exit_point = self.ql.exit_point
self.exit_point = self.load_address + len(self.ql.code)
# if exit point was set explicitly, override the default one
if self.ql.exit_point is not None:
self.exit_point = self.ql.exit_point
# if heap info is provided in profile, create heap
heap_base = self.profile.getint('CODE', 'heap_address', fallback=None)
heap_size = self.profile.getint('CODE', 'heap_size', fallback=None)
if heap_base is not None and heap_size is not None:
self.heap = QlMemoryHeap(self.ql, heap_base, heap_base + heap_size)
self.ql.emu_start(self.entry_point, self.exit_point, self.ql.timeout, self.ql.count)
+40 -37
View File
@@ -1,23 +1,33 @@
#!/usr/bin/env python3
#
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
from typing import AnyStr, Optional, Union
from .mapper import QlFsMappedObject
ReadableBuffer = Union[bytes, bytearray, memoryview]
# Open a file as a Disk
# host_path: The file path on the host machine.
# drive_path: The drive path on the emulated system. e.g. /dev/sda \\.\PHYSICALDRIVE0 0x80
#
#
# Note: CHS and LBA support is very limited since a raw file doesn't contain enough information.
# We simply assume that it is a disk with 1 head, 1 cylinder and (filesize/512) sectors.
#
# See: https://en.wikipedia.org/wiki/Cylinder-head-sector
# https://en.wikipedia.org/wiki/Logical_block_addressing
# http://www.uruk.org/orig-grub/PC_partitioning.txt
class QlDisk(QlFsMappedObject):
def __init__(self, host_path, drive_path, n_heads=1, n_cylinders=1, sector_size=512):
self._host_path = host_path
# 512 bytes/sector
# 63 sectors/track
# 255 heads (tracks/cylinder)
# 1024 cylinders
def __init__(self, host_path: AnyStr, drive_path, n_cylinders: int = 1, n_heads: int = 1, sector_size: int = 512):
self._drive_path = drive_path
self._fp = open(host_path, "rb+")
self._n_heads = n_heads
@@ -25,7 +35,7 @@ class QlDisk(QlFsMappedObject):
self._sector_size = sector_size
self.lseek(0, 2)
self._filesize = self.tell()
self._n_sectors = (self._filesize - 1)// self.sector_size + 1
self._n_sectors = (self._filesize - 1) // self.sector_size + 1
def __del__(self):
if not self.fp.closed:
@@ -51,50 +61,43 @@ class QlDisk(QlFsMappedObject):
def sector_size(self):
return self._sector_size
@property
def host_path(self):
return self._host_path
@property
def drive_path(self):
return self._drive_path
@property
def fp(self):
return self._fp
# Methods from FsMappedObject
def read(self, l):
return self.fp.read(l)
def write(self, bs):
return self.fp.write(bs)
def read(self, size: Optional[int]) -> bytes:
return self.fp.read(size)
def lseek(self, offset, origin):
def write(self, buffer: ReadableBuffer) -> int:
return self.fp.write(buffer)
def lseek(self, offset: int, origin: int) -> int:
return self.fp.seek(offset, origin)
def tell(self):
def tell(self) -> int:
return self.fp.tell()
def close(self):
return self.fp.close()
def close(self) -> None:
self.fp.close()
# Methods for QlDisk
def lba(self, cylinder, head, sector):
def lba(self, cylinder: int, head: int, sector: int) -> int:
return (cylinder * self.n_heads + head) * self._n_sectors + sector - 1
def read_sectors(self, lba, cnt):
def read_sectors(self, lba: int, cnt: int) -> bytes:
self.lseek(self.sector_size * lba, 0)
return self.read(self.sector_size*cnt)
def read_chs(self, cylinder, head, sector, cnt):
return self.read(self.sector_size * cnt)
def read_chs(self, cylinder: int, head: int, sector: int, cnt: int) -> bytes:
return self.read_sectors(self.lba(cylinder, head, sector), cnt)
def write_sectors(self, lba, cnt, buffer):
if len(buffer) > self.sector_size * cnt:
buffer = buffer[:self.sector_size*cnt]
def write_sectors(self, lba: int, cnt: int, buffer: ReadableBuffer) -> int:
buffer = memoryview(buffer)
self.lseek(self.sector_size * lba, 0)
return self.write(buffer)
def write_chs(self, cylinder, head, sector, cnt, buffer):
return self.write_sectors(self.lba(cylinder, head, sector), cnt, buffer)
return self.write(buffer[:self.sector_size * cnt])
def write_chs(self, cylinder: int, head: int, sector: int, cnt: int, buffer: ReadableBuffer):
return self.write_sectors(self.lba(cylinder, head, sector), cnt, buffer)
+39 -5
View File
@@ -9,11 +9,6 @@ from qiling import Qiling
from .. import utils
# exit
def __leaf_4c(ql: Qiling):
ql.log.info("Program terminated gracefully")
ql.emu_stop()
# write a character to screen
def __leaf_02(ql: Qiling):
ch = ql.arch.regs.dl
@@ -131,6 +126,45 @@ def __leaf_43(ql: Qiling):
ql.arch.regs.cx = 0xffff
ql.os.clear_cf()
def __leaf_48(ql: Qiling):
"""Allocate memory.
"""
size = ql.arch.regs.bx * 0x10
# announce it but do not do anything really
ql.log.debug(f'allocating memory block at {addr:#06x} to {size:#x} bytes')
# success
ql.os.clear_cf()
def __leaf_49(ql: Qiling):
"""Deallocate memory.
"""
...
def __leaf_4a(ql: Qiling):
"""Modify memory allocation.
"""
addr = ql.arch.regs.es
size = ql.arch.regs.bx * 0x10
# announce it but do not do anything really
ql.log.debug(f'resizing memory block at {addr:#06x} to {size:#x} bytes')
# success
ql.os.clear_cf()
def __leaf_4c(ql: Qiling):
ql.log.info("Program terminated gracefully")
ql.emu_stop()
def handler(ql: Qiling):
ah = ql.arch.regs.ah
+55 -23
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
#
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
@@ -53,31 +53,63 @@ def hook_mcount(ql, address, params):
return 0
@linux_kernel_api(params={
"Ptr": POINTER
})
def hook___x86_indirect_thunk_rax(ql, address, params):
return 0
def __x86_indirect_thunk(ql: Qiling, dest: int):
ql.log.debug('retpoline to %#010x', dest)
ql.arch.regs.arch_pc = dest
# using passthru as a hack to avoid syscall handler overwrite instruction pointer
@linux_kernel_api(passthru=True)
def hook___x86_indirect_thunk_rax(ql: Qiling, address: int, params):
__x86_indirect_thunk(ql, ql.arch.regs.rax)
@linux_kernel_api(params={
"Ptr": POINTER
})
def hook__copy_to_user(ql, address, params):
return 0
@linux_kernel_api(params={
"Ptr": POINTER
})
def hook__copy_from_user(ql, address, params):
return 0
@linux_kernel_api(params={
"Ptr": POINTER
})
@linux_kernel_api(passthru=True)
def hook___x86_indirect_thunk_r14(ql, address, params):
__x86_indirect_thunk(ql, ql.arch.regs.r14)
@linux_kernel_api(params={
"ubuf": POINTER,
"kbuf": POINTER,
"count": SIZE_T
})
def hook__copy_to_user(ql: Qiling, address: int, params) -> int:
ubuf = params['ubuf']
kbuf = params['kbuf']
count = params['count']
# if user-mode buffer is not available, fail
# TODO: also fail if destination is not writeable
if not ql.mem.is_mapped(ubuf, count):
return count
data = ql.mem.read(kbuf, count)
ql.mem.write(ubuf, data)
return 0
@linux_kernel_api(params={
"kbuf": POINTER,
"ubuf": POINTER,
"count": SIZE_T
})
def hook__copy_from_user(ql: Qiling, address: int, params) -> int:
ubuf = params['ubuf']
kbuf = params['kbuf']
count = params['count']
# if user-mode buffer is not available, fail
# TODO: also fail if source is not readable
if not ql.mem.is_mapped(ubuf, count):
return count
data = ql.mem.read(ubuf, count)
ql.mem.write(kbuf, data)
return 0
+99 -79
View File
@@ -6,15 +6,32 @@
import bisect
import os
import re
from typing import Any, Callable, Iterator, List, Mapping, Optional, Pattern, Sequence, Tuple, Union
from typing import Any, Callable, Dict, Iterator, List, Mapping, Optional, Pattern, Protocol, Sequence, Tuple, Union
from unicorn import UC_PROT_NONE, UC_PROT_READ, UC_PROT_WRITE, UC_PROT_EXEC, UC_PROT_ALL
from qiling import Qiling
from qiling.exception import *
# tuple: range start, range end, permissions mask, range label, is mmio?
MapInfoEntry = Tuple[int, int, int, str, bool]
class QlMmioHandler(Protocol):
"""A simple MMIO handler boilerplate that can be used to implement memory mapped devices.
This should be extended to implement mapped devices state machines. Note that the read and write
methods are optional, where their existance indicates whether the device supports the corresponding
operation. That is, an unimplemented method means the corresponding operation will be silently
dropped.
"""
def read(self, ql: Qiling, offset: int, size: int) -> int:
...
def write(self, ql: Qiling, offset: int, size: int, value: int) -> None:
...
# tuple: range start, range end, permissions mask, range label, mmio hander object (if mmio range)
MapInfoEntry = Tuple[int, int, int, str, Optional[QlMmioHandler]]
MmioReadCallback = Callable[[Qiling, int, int], int]
MmioWriteCallback = Callable[[Qiling, int, int, int], None]
@@ -29,7 +46,6 @@ class QlMemoryManager:
def __init__(self, ql: Qiling, pagesize: int = 0x1000):
self.ql = ql
self.map_info: List[MapInfoEntry] = []
self.mmio_cbs = {}
bit_stuff = {
64: (1 << 64) - 1,
@@ -48,6 +64,31 @@ class QlMemoryManager:
# make sure pagesize is a power of 2
assert self.pagesize & (self.pagesize - 1) == 0, 'pagesize has to be a power of 2'
self._packers = {
(1, True): ql.pack8s,
(2, True): ql.pack16s,
(4, True): ql.pack32s,
(8, True): ql.pack64s,
(1, False): ql.pack8,
(2, False): ql.pack16,
(4, False): ql.pack32,
(8, False): ql.pack64
}
self._unpackers = {
(1, True): ql.unpack8s,
(2, True): ql.unpack16s,
(4, True): ql.unpack32s,
(8, True): ql.unpack64s,
(1, False): ql.unpack8,
(2, False): ql.unpack16,
(4, False): ql.unpack32,
(8, False): ql.unpack64
}
def __read_string(self, addr: int) -> str:
ret = bytearray()
c = self.read(addr, 1)
@@ -80,7 +121,7 @@ class QlMemoryManager:
self.__write_string(addr, value, encoding)
def add_mapinfo(self, mem_s: int, mem_e: int, mem_p: int, mem_info: str, is_mmio: bool = False):
def add_mapinfo(self, mem_s: int, mem_e: int, mem_p: int, mem_info: str, mmio_ctx: Optional[QlMmioHandler] = None):
"""Add a new memory range to map.
Args:
@@ -88,10 +129,10 @@ class QlMemoryManager:
mem_e: memory range end
mem_p: permissions mask
mem_info: map entry label
is_mmio: memory range is mmio
mmio_ctx: mmio handler object; if specified the range will be treated as mmio
"""
bisect.insort(self.map_info, (mem_s, mem_e, mem_p, mem_info, is_mmio))
bisect.insort(self.map_info, (mem_s, mem_e, mem_p, mem_info, mmio_ctx))
def del_mapinfo(self, mem_s: int, mem_e: int):
"""Subtract a memory range from map.
@@ -105,13 +146,13 @@ class QlMemoryManager:
def __split_overlaps():
for idx in overlap_ranges:
lbound, ubound, perms, label, is_mmio = self.map_info[idx]
lbound, ubound, perms, label, mmio_ctx = self.map_info[idx]
if lbound < mem_s:
yield (lbound, mem_s, perms, label, is_mmio)
yield (lbound, mem_s, perms, label, mmio_ctx)
if mem_e < ubound:
yield (mem_e, ubound, perms, label, is_mmio)
yield (mem_e, ubound, perms, label, mmio_ctx)
# indices of first and last overlapping ranges. since map info is always
# sorted, we know that all overlapping rages are consecutive, so i1 > i0
@@ -129,27 +170,30 @@ class QlMemoryManager:
for entry in new_entries:
bisect.insort(self.map_info, entry)
def change_mapinfo(self, mem_s: int, mem_e: int, mem_p: Optional[int] = None, mem_info: Optional[str] = None):
tmp_map_info: Optional[MapInfoEntry] = None
info_idx: int = -1
def change_mapinfo(self, mem_s: int, mem_e: int, *, new_perms: Optional[int] = None, new_info: Optional[str] = None) -> None:
if new_perms is None and new_info is None:
# nothing to do
return
for idx, map_info in enumerate(self.map_info):
if mem_s >= map_info[0] and mem_e <= map_info[1]:
tmp_map_info = map_info
info_idx = idx
break
if tmp_map_info is None:
try:
# locate the map info entry to change
entry = next(entry for entry in self.map_info if mem_s >= entry[0] and mem_e <= entry[1])
except StopIteration:
self.ql.log.error(f'Cannot change mapinfo at {mem_s:#08x}-{mem_e:#08x}')
return
if mem_p is not None:
self.del_mapinfo(mem_s, mem_e)
self.add_mapinfo(mem_s, mem_e, mem_p, mem_info if mem_info else tmp_map_info[3])
return
_, _, perms, info, mmio_ctx = entry
if mem_info is not None:
self.map_info[info_idx] = (tmp_map_info[0], tmp_map_info[1], tmp_map_info[2], mem_info, tmp_map_info[4])
# caller wants to change perms?
if new_perms is not None:
perms = new_perms
# caller wants to change info?
if new_info is not None:
info = new_info
self.del_mapinfo(mem_s, mem_e)
self.add_mapinfo(mem_s, mem_e, perms, info, mmio_ctx)
def get_mapinfo(self) -> Sequence[Tuple[int, int, str, str, str]]:
"""Get memory map info.
@@ -168,18 +212,18 @@ class QlMemoryManager:
return ''.join(val if idx & ps else '-' for idx, val in perms_d.items())
def __process(lbound: int, ubound: int, perms: int, label: str, is_mmio: bool) -> Tuple[int, int, str, str, str]:
perms_str = __perms_mapping(perms)
def __process(entry: MapInfoEntry) -> Tuple[int, int, str, str, str]:
lbound, ubound, perms, label, mmio_ctx = entry
if hasattr(self.ql, 'loader'):
image = self.ql.loader.find_containing_image(lbound)
container = image.path if image and not is_mmio else ''
container = image.path if image and mmio_ctx is None else ''
else:
container = ''
return (lbound, ubound, perms_str, label, container)
return (lbound, ubound, __perms_mapping(perms), label, container)
return tuple(__process(*entry) for entry in self.map_info)
return tuple(__process(entry) for entry in self.map_info)
def get_formatted_mapinfo(self) -> Sequence[str]:
"""Get memory map info in a nicely formatted table.
@@ -270,12 +314,13 @@ class QlMemoryManager:
"mmio" : []
}
for lbound, ubound, perm, label, is_mmio in self.map_info:
if is_mmio:
mem_dict['mmio'].append((lbound, ubound, perm, label, *self.mmio_cbs[(lbound, ubound)]))
for lbound, ubound, perm, label, mmio_ctx in self.map_info:
if mmio_ctx is None:
key, data = 'ram', bytes(self.read(lbound, ubound - lbound))
else:
data = self.read(lbound, ubound - lbound)
mem_dict['ram'].append((lbound, ubound, perm, label, bytes(data)))
key, data = 'mmio', mmio_ctx
mem_dict[key].append((lbound, ubound, perm, label, data))
return mem_dict
@@ -294,12 +339,12 @@ class QlMemoryManager:
self.ql.log.debug(f'writing {len(data):#x} bytes at {lbound:#08x}')
self.write(lbound, data)
for lbound, ubound, perms, label, read_cb, write_cb in mem_dict['mmio']:
for lbound, ubound, perms, label, handler in mem_dict['mmio']:
self.ql.log.debug(f"restoring mmio range: {lbound:#08x} {ubound:#08x} {label}")
size = ubound - lbound
if not self.is_mapped(lbound, size):
self.map_mmio(lbound, size, read_cb, write_cb, info=label)
self.map_mmio(lbound, size, handler, label)
def read(self, addr: int, size: int) -> bytearray:
"""Read bytes from memory.
@@ -328,22 +373,12 @@ class QlMemoryManager:
if not size:
size = self.ql.arch.pointersize
__unpack = ({
1: self.ql.unpack8s,
2: self.ql.unpack16s,
4: self.ql.unpack32s,
8: self.ql.unpack64s
} if signed else {
1: self.ql.unpack8,
2: self.ql.unpack16,
4: self.ql.unpack32,
8: self.ql.unpack64
}).get(size)
if __unpack is None:
try:
_unpack = self._unpackers[(size, signed)]
except KeyError:
raise QlErrorStructConversion(f"Unsupported pointer size: {size}")
return __unpack(self.read(addr, size))
return _unpack(self.read(addr, size))
def write(self, addr: int, data: bytes) -> None:
"""Write bytes to a memory.
@@ -369,22 +404,12 @@ class QlMemoryManager:
if not size:
size = self.ql.arch.pointersize
__pack = ({
1: self.ql.pack8s,
2: self.ql.pack16s,
4: self.ql.pack32s,
8: self.ql.pack64s
} if signed else {
1: self.ql.pack8,
2: self.ql.pack16,
4: self.ql.pack32,
8: self.ql.pack64
}).get(size)
if __pack is None:
try:
_pack = self._packers[(size, signed)]
except KeyError:
raise QlErrorStructConversion(f"Unsupported pointer size: {size}")
self.write(addr, __pack(value))
self.write(addr, _pack(value))
def search(self, needle: Union[bytes, Pattern[bytes]], begin: Optional[int] = None, end: Optional[int] = None) -> List[int]:
"""Search for a sequence of bytes in memory.
@@ -408,7 +433,7 @@ class QlMemoryManager:
assert begin < end, 'search arguments do not make sense'
# narrow the search down to relevant ranges; mmio ranges are excluded due to potential read side effects
ranges = [(max(begin, lbound), min(ubound, end)) for lbound, ubound, _, _, is_mmio in self.map_info if not (end < lbound or ubound < begin or is_mmio)]
ranges = [(max(begin, lbound), min(ubound, end)) for lbound, ubound, _, _, mmio_ctx in self.map_info if not (end < lbound or ubound < begin or mmio_ctx is not None)]
results = []
# if needle is a bytes sequence use it verbatim, not as a pattern
@@ -434,9 +459,6 @@ class QlMemoryManager:
self.del_mapinfo(addr, addr + size)
self.ql.uc.mem_unmap(addr, size)
if (addr, addr + size) in self.mmio_cbs:
del self.mmio_cbs[(addr, addr+size)]
def unmap_between(self, mem_s: int, mem_e: int) -> None:
"""Reclaim any allocated memory region within the specified range.
@@ -595,7 +617,7 @@ class QlMemoryManager:
aligned_size = self.align_up((addr & (self.pagesize - 1)) + size)
self.ql.uc.mem_protect(aligned_address, aligned_size, perms)
self.change_mapinfo(aligned_address, aligned_address + aligned_size, perms)
self.change_mapinfo(aligned_address, aligned_address + aligned_size, new_perms=perms)
def map(self, addr: int, size: int, perms: int = UC_PROT_ALL, info: Optional[str] = None):
"""Map a new memory range.
@@ -617,17 +639,17 @@ class QlMemoryManager:
raise QlMemoryMappedError('Requested memory is unavailable')
self.ql.uc.mem_map(addr, size, perms)
self.add_mapinfo(addr, addr + size, perms, info or '[mapped]', is_mmio=False)
self.add_mapinfo(addr, addr + size, perms, info or '[mapped]', None)
def map_mmio(self, addr: int, size: int, read_cb: Optional[MmioReadCallback], write_cb: Optional[MmioWriteCallback], info: str = '[mmio]'):
def map_mmio(self, addr: int, size: int, handler: QlMmioHandler, info: str = '[mmio]'):
# TODO: mmio memory overlap with ram? Is that possible?
# TODO: Can read_cb or write_cb be None? How uc handle that access?
prot = UC_PROT_NONE
if read_cb:
if hasattr(handler, 'read'):
prot |= UC_PROT_READ
if write_cb:
if hasattr(handler, 'write'):
prot |= UC_PROT_WRITE
# generic mmio read wrapper
@@ -642,10 +664,8 @@ class QlMemoryManager:
cb(self.ql, offset, size, value)
self.ql.uc.mmio_map(addr, size, __mmio_read, read_cb, __mmio_write, write_cb)
self.add_mapinfo(addr, addr + size, prot, info, is_mmio=True)
self.mmio_cbs[(addr, addr + size)] = (read_cb, write_cb)
self.ql.uc.mmio_map(addr, size, __mmio_read, handler.read, __mmio_write, handler.write)
self.add_mapinfo(addr, addr + size, prot, info, handler)
class Chunk:
+1
View File
@@ -89,6 +89,7 @@ class QlOs:
if self.ql.code:
# this shellcode entrypoint does not work for windows
# windows shellcode entry point will comes from pe loader
self.load_address = self.profile.getint('CODE', 'load_address')
self.entry_point = self.profile.getint('CODE', 'entry_point')
self.code_ram_size = self.profile.getint('CODE', 'ram_size')
+1
View File
@@ -14,6 +14,7 @@ from .prctl import *
from .ptrace import *
from .random import *
from .resource import *
from .rseq import *
from .sched import *
from .select import *
from .sendfile import *
+13
View File
@@ -0,0 +1,13 @@
#!/usr/bin/env python3
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
from qiling import Qiling
def ql_syscall_rseq(ql: Qiling, rseq: int, rseq_len: int, flags: int, sig: int):
# indicate rseq is not supported by this kernel
# return -ENOSYS
return 0
+2
View File
@@ -415,6 +415,8 @@ def ql_syscall_read(ql: Qiling, fd: int, buf: int, length: int):
try:
data = f.read(length)
except IsADirectoryError:
return -EISDIR
except ConnectionError:
ql.log.debug('read failed due to a connection error')
return -EIO
+5 -9
View File
@@ -10,6 +10,10 @@ from .ProcessorBind import *
from .UefiBaseType import *
from .UefiMultiPhase import *
from .protocols.EfiSimpleTextInProtocol import EFI_SIMPLE_TEXT_INPUT_PROTOCOL
from .protocols.EfiSimpleTextOutProtocol import EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL
# definitions for EFI_TIME.Daylight
EFI_TIME_ADJUST_DAYLIGHT = (1 << 1)
EFI_TIME_IN_DAYLIGHT = (1 << 2)
@@ -223,14 +227,6 @@ class EFI_CONFIGURATION_TABLE(STRUCT):
('VendorTable', PTR(VOID)),
]
# TODO: to be implemented
# @see: MdePkg\Include\Protocol\SimpleTextIn.h
EFI_SIMPLE_TEXT_INPUT_PROTOCOL = STRUCT
# TODO: to be implemented
# @see: MdePkg\Include\Protocol\SimpleTextOut.h
EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL = STRUCT
class EFI_SYSTEM_TABLE(STRUCT):
_pack_ = 8
@@ -264,4 +260,4 @@ __all__ = [
'EFI_DEVICE_PATH_PROTOCOL',
'EFI_OPEN_PROTOCOL_INFORMATION_ENTRY',
'EFI_IMAGE_UNLOAD'
]
]
+4 -4
View File
@@ -1,14 +1,14 @@
#!/usr/bin/env python3
#
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
from typing import Any, Mapping
from typing import Any, Mapping, Optional
from qiling import Qiling
from qiling.const import QL_INTERCEPT
def dxeapi(params: Mapping[str, Any] = {}):
def dxeapi(params: Optional[Mapping[str, Any]] = None, passthru: bool = False):
def decorator(func):
def wrapper(ql: Qiling):
pc = ql.arch.regs.arch_pc
@@ -18,7 +18,7 @@ def dxeapi(params: Mapping[str, Any] = {}):
onenter = ql.os.user_defined_api[QL_INTERCEPT.ENTER].get(fname)
onexit = ql.os.user_defined_api[QL_INTERCEPT.EXIT].get(fname)
return ql.os.call(pc, f, params, onenter, onexit)
return ql.os.call(pc, f, params or {}, onenter, onexit, passthru)
return wrapper
@@ -0,0 +1,56 @@
#!/usr/bin/env python3
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
from qiling.os.const import *
from qiling.os.uefi.fncc import dxeapi
from qiling.os.uefi.utils import *
from qiling.os.uefi.ProcessorBind import *
from qiling.os.uefi.UefiBaseType import EFI_STATUS, EFI_EVENT
# @see: MdePkg/Include/Protocol/SimpleTextIn.h
class EFI_INPUT_KEY(STRUCT):
_fields_ = [
('ScanCode', UINT16),
('UnicodeChar', CHAR16)
]
class EFI_SIMPLE_TEXT_INPUT_PROTOCOL(STRUCT):
EFI_SIMPLE_TEXT_INPUT_PROTOCOL = STRUCT
_fields_ = [
('Reset', FUNCPTR(EFI_STATUS, PTR(EFI_SIMPLE_TEXT_INPUT_PROTOCOL), BOOLEAN)),
('ReadKeyStroke', FUNCPTR(EFI_STATUS, PTR(EFI_SIMPLE_TEXT_INPUT_PROTOCOL), PTR(EFI_INPUT_KEY))),
('WaitForKey', EFI_EVENT)
]
@dxeapi(params={
"This": POINTER, # IN PTR(EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL)
"ExtendedVerification": BOOL # IN BOOLEAN
})
def hook_Input_Reset(ql: Qiling, address: int, params):
pass
@dxeapi(params={
"This": POINTER, # IN PTR(EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL)
"Key": POINTER # OUT PTR(EFI_INPUT_KEY)
})
def hook_Read_Key_Stroke(ql: Qiling, address: int, params):
pass
def initialize(ql: Qiling, gIP: int):
descriptor = {
'struct': EFI_SIMPLE_TEXT_INPUT_PROTOCOL,
'fields': (
('Reset', hook_Input_Reset),
('ReadKeyStroke', hook_Read_Key_Stroke),
('WaitForKey', None)
)
}
instance = init_struct(ql, gIP, descriptor)
instance.save_to(ql.mem, gIP)
@@ -0,0 +1,128 @@
#!/usr/bin/env python3
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
from qiling.os.const import *
from qiling.os.uefi.fncc import dxeapi
from qiling.os.uefi.utils import *
from qiling.os.uefi.ProcessorBind import *
from qiling.os.uefi.UefiBaseType import EFI_STATUS
# @see: MdePkg/Include/Protocol/SimpleTextOut.h
class SIMPLE_TEXT_OUTPUT_MODE(STRUCT):
_fields_ = [
("MaxMode", INT32),
("Mode", INT32),
("Attribute", INT32),
("CursorColumn", INT32),
("CursorRow", INT32),
("CursorVisible", BOOLEAN),
]
class EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL(STRUCT):
EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL = STRUCT
_fields_ = [
("Reset", FUNCPTR(EFI_STATUS, PTR(EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL), BOOLEAN)),
("OutputString", FUNCPTR(EFI_STATUS, PTR(EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL), PTR(CHAR16))),
("TestString", FUNCPTR(EFI_STATUS, PTR(EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL), PTR(CHAR16))),
("QueryMode", FUNCPTR(EFI_STATUS, PTR(EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL), UINTN, PTR(UINTN), PTR(UINTN))),
("SetMode", FUNCPTR(EFI_STATUS, PTR(EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL), UINTN)),
("SetAttribute", FUNCPTR(EFI_STATUS, PTR(EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL), UINTN)),
("ClearScreen", FUNCPTR(EFI_STATUS, PTR(EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL))),
("SetCursorPosition", FUNCPTR(EFI_STATUS, PTR(EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL), UINTN, UINTN)),
("EnableCursor", FUNCPTR(EFI_STATUS, PTR(EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL), BOOLEAN)),
("Mode", PTR(SIMPLE_TEXT_OUTPUT_MODE))
]
@dxeapi(params={
"This": POINTER, # IN PTR(EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL)
"ExtendedVerification": BOOL # IN BOOLEAN
})
def hook_TextReset(ql: Qiling, address: int, params):
pass
@dxeapi(params={
"This": POINTER, # IN PTR(EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL)
"String": WSTRING # IN PTR(CHAR16)
})
def hook_OutputString(ql: Qiling, address: int, params):
print(params['String'])
return EFI_SUCCESS
@dxeapi(params={
"This": POINTER, # IN PTR(EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL)
"String": WSTRING # IN PTR(CHAR16)
})
def hook_TestString(ql: Qiling, address: int, params):
pass
@dxeapi(params={
"This": POINTER, # IN PTR(EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL)
"ModeNumber": ULONGLONG, # IN UINTN
"Columns": POINTER, # OUT PTR(UINTN)
"Rows": POINTER # OUT PTR(UINTN)
})
def hook_QueryMode(ql: Qiling, address: int, params):
pass
@dxeapi(params={
"This": POINTER, # IN PTR(EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL)
"ModeNumber": ULONGLONG # IN UINTN
})
def hook_SetMode(ql: Qiling, address: int, params):
pass
@dxeapi(params={
"This": POINTER, # IN PTR(EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL)
"Attribute": ULONGLONG # IN UINTN
})
def hook_SetAttribute(ql: Qiling, address: int, params):
pass
@dxeapi(params={
"This": POINTER # IN PTR(EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL)
})
def hook_ClearScreen(ql: Qiling, address: int, params):
pass
@dxeapi(params={
"This": POINTER, # IN PTR(EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL)
"Column": ULONGLONG, # IN UINTN
"Row": ULONGLONG # IN UINTN
})
def hook_SetCursorPosition(ql: Qiling, address: int, params):
pass
@dxeapi(params={
"This": POINTER, # IN PTR(EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL)
"Visible": BOOL # IN BOOLEAN
})
def hook_EnableCursor(ql: Qiling, address: int, params):
pass
def initialize(ql: Qiling, base: int):
descriptor = {
'struct': EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL,
'fields': (
('Reset', hook_TextReset),
('OutputString', hook_OutputString),
('TestString', hook_TestString),
('QueryMode', hook_QueryMode),
('SetMode', hook_SetMode),
('SetAttribute', hook_SetAttribute),
('ClearScreen', hook_ClearScreen),
('SetCursorPosition', hook_SetCursorPosition),
('EnableCursor', hook_EnableCursor),
('Mode', None)
)
}
instance = init_struct(ql, base, descriptor)
instance.save_to(ql.mem, base)
+67 -39
View File
@@ -3,58 +3,81 @@
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
from qiling import Qiling
from __future__ import annotations
from typing import TYPE_CHECKING
from qiling.os.uefi import bs, rt, ds
from qiling.os.uefi.context import UefiContext
from qiling.os.uefi.utils import install_configuration_table
from qiling.os.uefi.UefiSpec import EFI_SYSTEM_TABLE, EFI_BOOT_SERVICES, EFI_RUNTIME_SERVICES
from qiling.os.uefi.UefiSpec import EFI_SYSTEM_TABLE, EFI_SIMPLE_TEXT_INPUT_PROTOCOL, EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL, EFI_BOOT_SERVICES, EFI_RUNTIME_SERVICES
import qiling.os.uefi.protocols.EfiSimpleTextInProtocol as txt_in
import qiling.os.uefi.protocols.EfiSimpleTextOutProtocol as txt_out
if TYPE_CHECKING:
from qiling import Qiling
# static mem layout:
#
# +-- EFI_SYSTEM_TABLE ---------+
# | |
# | ... |
# | RuntimeServices* -> (1) |
# | BootServices* -> (2) |
# | NumberOfTableEntries |
# | ConfigurationTable* -> (4) |
# +-----------------------------+
# (1) +-- EFI_RUNTIME_SERVICES -----+
# | |
# | ... |
# +-----------------------------+
# (2) +-- EFI_BOOT_SERVICES --------+
# | |
# | ... |
# +-----------------------------+
# (3) +-- EFI_DXE_SERVICES ---------+
# | |
# | ... |
# +-----------------------------+
# (4) +-- EFI_CONFIGURATION_TABLE --+ of HOB_LIST
# | VendorGuid |
# | VendorTable* -> (5) |
# +-----------------------------+
# +-- EFI_CONFIGURATION_TABLE --+ of DXE_SERVICE_TABLE
# | VendorGuid |
# | VendorTable* -> (3) |
# +-----------------------------+
# +-- EFI_SYSTEM_TABLE -----------------+
# | |
# | ... |
# | ConIn* -> (1) |
# | ConOut* -> (2) |
# | RuntimeServices* -> (3) |
# | BootServices* -> (4) |
# | NumberOfTableEntries |
# | ConfigurationTable* -> (6) |
# +-------------------------------------+
# (1) +-- EFI_SIMPLE_TEXT_INPUT_PROTOCOL ---+
# | |
# | ... |
# +-------------------------------------+
# (2) +-- EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL --+
# | |
# | ... |
# +-------------------------------------+
# (3) +-- EFI_RUNTIME_SERVICES -------------+
# | |
# | ... |
# +-------------------------------------+
# (4) +-- EFI_BOOT_SERVICES ----------------+
# | |
# | ... |
# +-------------------------------------+
# (5) +-- EFI_DXE_SERVICES -----------------+
# | |
# | ... |
# +-------------------------------------+
# (6) +-- EFI_CONFIGURATION_TABLE ----------+ of HOB_LIST
# | VendorGuid |
# | VendorTable* -> (7) |
# +-------------------------------------+
# +-- EFI_CONFIGURATION_TABLE ----------+ of DXE_SERVICE_TABLE
# | VendorGuid |
# | VendorTable* -> (5) |
# +-------------------------------------+
#
# ... the remainder of the chunk may be used for additional EFI_CONFIGURATION_TABLE entries
#
# dynamically allocated (context.conf_table_data_ptr):
#
# (5) +-- VOID* --------------------+
# | ... |
# +-----------------------------+
# (7) +-- VOID* ----------------------------+
# | ... |
# +-------------------------------------+
def initialize(ql: Qiling, context: UefiContext, gST: int):
ql.loader.gST = gST
gBS = gST + EFI_SYSTEM_TABLE.sizeof() # boot services
gRT = gBS + EFI_BOOT_SERVICES.sizeof() # runtime services
gDS = gRT + EFI_RUNTIME_SERVICES.sizeof() # dxe services
cfg = gDS + ds.EFI_DXE_SERVICES.sizeof() # configuration tables array
sti = gST + EFI_SYSTEM_TABLE.sizeof() # input protocols
sto = sti + EFI_SIMPLE_TEXT_INPUT_PROTOCOL.sizeof() # output protocols
gRT = sto + EFI_SIMPLE_TEXT_OUTPUT_PROTOCOL.sizeof() # runtime services
gBS = gRT + EFI_RUNTIME_SERVICES.sizeof() # boot services
gDS = gBS + EFI_BOOT_SERVICES.sizeof() # dxe services
cfg = gDS + ds.EFI_DXE_SERVICES.sizeof() # configuration tables array
ql.log.info(f'Global tables:')
ql.log.info(f' | gST {gST:#010x}')
@@ -63,11 +86,16 @@ def initialize(ql: Qiling, context: UefiContext, gST: int):
ql.log.info(f' | gDS {gDS:#010x}')
ql.log.info(f'')
txt_in.initialize(ql, sti)
txt_out.initialize(ql, sto)
bs.initialize(ql, gBS)
rt.initialize(ql, gRT)
ds.initialize(ql, gDS)
EFI_SYSTEM_TABLE(
ConIn = sti,
ConOut = sto,
RuntimeServices = gRT,
BootServices = gBS,
NumberOfTableEntries = 0,
@@ -79,4 +107,4 @@ def initialize(ql: Qiling, context: UefiContext, gST: int):
__all__ = [
'initialize'
]
]
+1
View File
@@ -1,6 +1,7 @@
[CODE]
# ram_size 0xa00000 is 10MB
ram_size = 0xa00000
load_address = 0x1000000
entry_point = 0x1000000
+1
View File
@@ -23,6 +23,7 @@ KI_USER_SHARED_DATA = 0x7ffe0000
[CODE]
# ram_size 0xa00000 is 10MB
ram_size = 0xa00000
load_address = 0x1000000
entry_point = 0x1000000
[KERNEL]
View File
+4
View File
@@ -0,0 +1,4 @@
[CODE]
load_address = 0x10000000
entry_point = 0x10000008
ram_size = 0xa00000
+2
View File
@@ -1,6 +1,8 @@
[CODE]
ram_size = 0xa00000
load_address = 0x80800000
entry_point = 0x80800000
heap_address = 0xa0000000
heap_size = 0x300000
+89 -12
View File
@@ -10,13 +10,17 @@ sys.path.append("..")
from qiling.core import Qiling
from qiling.const import QL_ARCH, QL_OS, QL_VERBOSE
from qiling.os.const import STRING
from qiling.os.const import STRING, POINTER, SIZE_T
class BlobTest(unittest.TestCase):
def test_uboot_arm(self):
def my_getenv(ql, *args, **kwargs):
env = {"ID": b"000000000000000", "ethaddr": b"11:22:33:44:55:66"}
def my_getenv(ql: Qiling):
env = {
"ID": b"000000000000000",
"ethaddr": b"11:22:33:44:55:66"
}
params = ql.os.resolve_fcall_params({'key': STRING})
value = env.get(params["key"], b"")
@@ -26,12 +30,23 @@ class BlobTest(unittest.TestCase):
ql.arch.regs.r0 = value_addr
ql.arch.regs.arch_pc = ql.arch.regs.lr
def check_password(ql, *args, **kwargs):
passwd_output = ql.mem.read(ql.arch.regs.r0, ql.arch.regs.r2)
passwd_input = ql.mem.read(ql.arch.regs.r1, ql.arch.regs.r2)
self.assertEqual(passwd_output, passwd_input)
def check_password(ql: Qiling):
params = ql.os.resolve_fcall_params({
'ptr1': POINTER, # points to real password
'ptr2': POINTER, # points to user provided password
'size': SIZE_T # comparison length
})
def partial_run_init(ql):
ptr1 = params['ptr1']
ptr2 = params['ptr2']
size = params['size']
real_password = ql.mem.read(ptr1, size)
user_password = ql.mem.read(ptr2, size)
self.assertSequenceEqual(real_password, user_password, seq_type=bytearray)
def partial_run_init(ql: Qiling):
# argv prepare
ql.arch.regs.arch_sp -= 0x30
arg0_ptr = ql.arch.regs.arch_sp
@@ -56,16 +71,78 @@ class BlobTest(unittest.TestCase):
ql = Qiling(code=uboot_code[0x40:], archtype=QL_ARCH.ARM, ostype=QL_OS.BLOB, profile="profiles/uboot_bin.ql", verbose=QL_VERBOSE.DEBUG)
image_base_addr = ql.loader.load_address
ql.hook_address(my_getenv, image_base_addr + 0x13AC0)
ql.hook_address(check_password, image_base_addr + 0x48634)
imgbase = ql.loader.images[0].base
ql.hook_address(my_getenv, imgbase + 0x13AC0)
ql.hook_address(check_password, imgbase + 0x48634)
partial_run_init(ql)
ql.run(image_base_addr + 0x486B4, image_base_addr + 0x48718)
ql.run(imgbase + 0x486B4, imgbase + 0x48718)
del ql
def test_blob_raw(self):
def run_checksum_emu(input_data_buffer: bytes) -> int:
"""
Callable function that takes input data buffer and returns the checksum.
"""
BASE_ADDRESS = 0x10000000
CHECKSUM_FUNC_ADDR = BASE_ADDRESS + 0x8
END_ADDRESS = 0x100000ba
DATA_ADDR = 0xa0000000
STACK_ADDR = 0xb0000000
with open("../examples/rootfs/blob/example_raw.bin", "rb") as f:
raw_code = f.read()
ql = Qiling(code=raw_code, archtype=QL_ARCH.ARM, ostype=QL_OS.BLOB, profile="profiles/blob_raw.ql", verbose=QL_VERBOSE.DEBUG, thumb=True)
input_data_len = len(input_data_buffer)
# Map memory for data and stack
ql.mem.map(STACK_ADDR, 0x2000)
ql.mem.map(DATA_ADDR, ql.mem.align_up(input_data_len + 0x100))
# Write input data
ql.mem.write(DATA_ADDR, input_data_buffer)
# Set up registers
ql.arch.regs.sp = STACK_ADDR + 0x2000 - 4
ql.arch.regs.r0 = DATA_ADDR
ql.arch.regs.r1 = input_data_len
ql.arch.regs.pc = CHECKSUM_FUNC_ADDR
ql.arch.regs.lr = 0xbebebebe
ql.run(begin=CHECKSUM_FUNC_ADDR, end=END_ADDRESS)
result = ql.arch.regs.r0
return result
def calculate_expected_checksum(input_data_buffer: bytes) -> int:
"""
Python implementation of the expected checksum calculation.
"""
input_data_len = len(input_data_buffer)
expected_checksum = 0
if input_data_len >= 1 and input_data_buffer[0] == 0xDE: # MAGIC_VALUE_1
for i in range(min(input_data_len, 4)):
expected_checksum += input_data_buffer[i]
expected_checksum += 0x10
elif input_data_len >= 2 and input_data_buffer[1] == 0xAD: # MAGIC_VALUE_2
for i in range(input_data_len):
expected_checksum ^= input_data_buffer[i]
expected_checksum += 0x20
else:
for i in range(input_data_len):
expected_checksum += input_data_buffer[i]
return expected_checksum & 0xFF
test_input = b"\x01\x02\x03\x04\x05"
self.assertEqual(run_checksum_emu(test_input), calculate_expected_checksum(test_input))
if __name__ == "__main__":
unittest.main()
+2 -2
View File
@@ -219,8 +219,8 @@ class PETest(unittest.TestCase):
fcall.writeParams(((DWORD, 0),))
# run until third stop
# TODO: Should stop at 0x10423, but for now just stop at 0x0001066a
amsint32.hook_address(hook_third_stop_address, 0x0001066a, stops)
# TODO: Should stop at 0x10423, but for now just stop at 0x10430
amsint32.hook_address(hook_third_stop_address, 0x10430, stops)
amsint32.run(begin=0x102D0)
self.assertTrue(stops[0])