qilingframework/qiling
1
0
Fork 0
mirror of https://github.com/qilingframework/qiling.git synced 2026-10-12 02:21:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1376 from elicn/periodic

Browse Source 
Periodic maintenance PR
This commit is contained in:
xwings
2023-09-26 17:17:18 +08:00
committed by GitHub
parent b5e8b53098 f4a9f85e42
commit 9b37c3b5cd
52 changed files with 1993 additions and 1392 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.editorconfig
+12
View File
@@ -0,0 +1,12 @@
# EditorConfig is awesome: https://editorconfig.org
# top-most EditorConfig file
root = true
[*.py]
charset = utf-8
end_of_line = lf
indent_size = 4
indent_style = space
insert_final_newline = true
trim_trailing_whitespace = true
qiling/arch/arch.py
+10 -5
View File
@@ -3,7 +3,8 @@
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
from abc import abstractmethod
from abc import ABC, abstractmethod
from typing import ClassVar, Optional
from unicorn import Uc
from unicorn.unicorn import UcContext
@@ -12,16 +13,20 @@ from keystone import Ks
from qiling import Qiling
from qiling.const import QL_ARCH, QL_ENDIAN
from .models import QL_CPU
from .register import QlRegisterManager
from .utils import QlArchUtils
class QlArch:
type: QL_ARCH
bits: int
class QlArch(ABC):
type: ClassVar[QL_ARCH]
bits: ClassVar[int]
def __init__(self, ql: Qiling):
def __init__(self, ql: Qiling, *, cputype: Optional[QL_CPU] = None):
self.ql = ql
self.cpu = cputype
self.utils = QlArchUtils(ql)
@property
qiling/arch/arm.py
+13 -6
View File
@@ -4,6 +4,7 @@
#
from functools import cached_property, lru_cache
from typing import Optional
from unicorn import Uc, UC_ARCH_ARM, UC_MODE_ARM, UC_MODE_THUMB, UC_MODE_BIG_ENDIAN
from capstone import Cs, CS_ARCH_ARM, CS_MODE_ARM, CS_MODE_THUMB, CS_MODE_BIG_ENDIAN
@@ -12,6 +13,7 @@ from keystone import Ks, KS_ARCH_ARM, KS_MODE_ARM, KS_MODE_THUMB, KS_MODE_BIG_EN
from qiling import Qiling
from qiling.arch.arch import QlArch
from qiling.arch import arm_const
from qiling.arch.models import ARM_CPU_MODEL
from qiling.arch.register import QlRegisterManager
from qiling.const import QL_ARCH, QL_ENDIAN
@@ -20,8 +22,8 @@ class QlArchARM(QlArch):
type = QL_ARCH.ARM
bits = 32
def __init__(self, ql: Qiling, endian: QL_ENDIAN, thumb: bool):
super().__init__(ql)
def __init__(self, ql: Qiling, *, cputype: Optional[ARM_CPU_MODEL], endian: QL_ENDIAN, thumb: bool):
super().__init__(ql, cputype=cputype)
self._init_endian = endian
self._init_thumb = thumb
@@ -32,13 +34,18 @@ class QlArchARM(QlArch):
def uc(self) -> Uc:
mode = UC_MODE_ARM
if self._init_endian == QL_ENDIAN.EB:
if self._init_endian is QL_ENDIAN.EB:
mode += UC_MODE_BIG_ENDIAN
if self._init_thumb:
mode += UC_MODE_THUMB
return Uc(UC_ARCH_ARM, mode)
obj = Uc(UC_ARCH_ARM, mode)
if self.cpu is not None:
obj.ctl_set_cpu_model(self.cpu.value)
return obj
@cached_property
def regs(self) -> QlRegisterManager:
@@ -82,7 +89,7 @@ class QlArchARM(QlArch):
mode = CS_MODE_ARM
if self.endian == QL_ENDIAN.EB:
if self.endian is QL_ENDIAN.EB:
mode += CS_MODE_BIG_ENDIAN
if self.is_thumb:
@@ -102,7 +109,7 @@ class QlArchARM(QlArch):
mode = KS_MODE_ARM
if self.endian == QL_ENDIAN.EB:
if self.endian is QL_ENDIAN.EB:
mode += KS_MODE_BIG_ENDIAN
if self.is_thumb:
qiling/arch/arm64.py
+12 -1
View File
@@ -4,13 +4,16 @@
#
from functools import cached_property
from typing import Optional
from unicorn import Uc, UC_ARCH_ARM64, UC_MODE_ARM
from capstone import Cs, CS_ARCH_ARM64, CS_MODE_ARM
from keystone import Ks, KS_ARCH_ARM64, KS_MODE_ARM
from qiling import Qiling
from qiling.arch.arch import QlArch
from qiling.arch import arm64_const
from qiling.arch.models import ARM64_CPU_MODEL
from qiling.arch.register import QlRegisterManager
from qiling.const import QL_ARCH, QL_ENDIAN
@@ -19,9 +22,17 @@ class QlArchARM64(QlArch):
type = QL_ARCH.ARM64
bits = 64
def __init__(self, ql: Qiling, *, cputype: Optional[ARM64_CPU_MODEL] = None):
super().__init__(ql, cputype=cputype)
@cached_property
def uc(self) -> Uc:
return Uc(UC_ARCH_ARM64, UC_MODE_ARM)
obj = Uc(UC_ARCH_ARM64, UC_MODE_ARM)
if self.cpu is not None:
obj.ctl_set_cpu_model(self.cpu.value)
return obj
@cached_property
def regs(self) -> QlRegisterManager:
qiling/arch/cortex_m.py
+13 -8
View File
@@ -5,6 +5,7 @@
from functools import cached_property
from contextlib import ContextDecorator
from typing import Optional
from unicorn import UC_ARCH_ARM, UC_MODE_ARM, UC_MODE_MCLASS, UC_MODE_THUMB
from capstone import Cs, CS_ARCH_ARM, CS_MODE_ARM, CS_MODE_MCLASS, CS_MODE_THUMB
@@ -13,6 +14,7 @@ from keystone import Ks, KS_ARCH_ARM, KS_MODE_ARM, KS_MODE_THUMB
from qiling import Qiling
from qiling.arch.arm import QlArchARM
from qiling.arch import cortex_m_const
from qiling.arch.models import ARM_CPU_MODEL
from qiling.arch.register import QlRegisterManager
from qiling.arch.cortex_m_const import IRQ, EXC_RETURN, CONTROL, EXCP
from qiling.const import QL_ARCH, QL_ENDIAN, QL_VERBOSE
@@ -66,12 +68,14 @@ class QlArchCORTEX_M(QlArchARM):
type = QL_ARCH.CORTEX_M
bits = 32
def __init__(self, ql: Qiling):
super().__init__(ql, endian=QL_ENDIAN.EL, thumb=True)
def __init__(self, ql: Qiling, *, cputype: Optional[ARM_CPU_MODEL] = None):
super().__init__(ql, cputype=cputype, endian=QL_ENDIAN.EL, thumb=True)
@cached_property
def uc(self):
return MultiTaskUnicorn(UC_ARCH_ARM, UC_MODE_ARM + UC_MODE_MCLASS + UC_MODE_THUMB, 10)
cpu = self.cpu and self.cpu.value
return MultiTaskUnicorn(UC_ARCH_ARM, UC_MODE_ARM + UC_MODE_MCLASS + UC_MODE_THUMB, cpu, 10)
@cached_property
def regs(self) -> QlRegisterManager:
@@ -97,18 +101,18 @@ class QlArchCORTEX_M(QlArchARM):
def endian(self) -> QL_ENDIAN:
return QL_ENDIAN.EL
def is_handler_mode(self):
def is_handler_mode(self) -> bool:
return self.regs.ipsr > 1
def using_psp(self):
def using_psp(self) -> bool:
return not self.is_handler_mode() and (self.regs.control & CONTROL.SPSEL) > 0
def init_context(self):
def init_context(self) -> None:
self.regs.lr = 0xffffffff
self.regs.msp = self.ql.mem.read_ptr(0x0)
self.regs.pc = self.ql.mem.read_ptr(0x4)
def unicorn_exception_handler(self, ql, intno):
def unicorn_exception_handler(self, ql: Qiling, intno: int):
forward_mapper = {
EXCP.UDEF : IRQ.HARD_FAULT, # undefined instruction
EXCP.SWI : IRQ.SVCALL, # software interrupt
@@ -139,8 +143,9 @@ class QlArchCORTEX_M(QlArchARM):
except IndexError:
raise QlErrorNotImplemented(f'Unhandled interrupt number ({intno})')
def interrupt_handler(self, ql, intno):
def interrupt_handler(self, ql: Qiling, intno: int):
basepri = self.regs.basepri & 0xf0
if basepri and basepri <= ql.hw.nvic.get_priority(intno):
return
qiling/arch/evm/evm.py
+16 -2
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
#
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
@@ -11,6 +11,7 @@ from qiling.arch.evm.vm.evm import QlArchEVMEmulator
from qiling.arch.evm.vm.message import Message
from qiling.const import *
class QlArchEVM(QlArch):
type = QL_ARCH.EVM
bits = 1
@@ -39,7 +40,19 @@ class QlArchEVM(QlArch):
@property
def uc(self):
return None
raise AttributeError
@property
def regs(self):
raise AttributeError
@property
def disassembler(self):
raise AttributeError
@property
def assembler(self):
raise AttributeError
@property
def endian(self) -> QL_ENDIAN:
@@ -49,6 +62,7 @@ class QlArchEVM(QlArch):
def __evm_run(self, code: Message):
return self.arch.run(code)
def monkeypatch_core_methods(ql):
"""Monkeypatch core methods for evm
"""
qiling/arch/mips.py
+4 -2
View File
@@ -4,6 +4,7 @@
#
from functools import cached_property
from typing import Optional
from unicorn import Uc, UC_ARCH_MIPS, UC_MODE_MIPS32, UC_MODE_BIG_ENDIAN, UC_MODE_LITTLE_ENDIAN
from capstone import Cs, CS_ARCH_MIPS, CS_MODE_MIPS32, CS_MODE_BIG_ENDIAN, CS_MODE_LITTLE_ENDIAN
@@ -12,6 +13,7 @@ from keystone import Ks, KS_ARCH_MIPS, KS_MODE_MIPS32, KS_MODE_BIG_ENDIAN, KS_MO
from qiling import Qiling
from qiling.arch.arch import QlArch
from qiling.arch import mips_const
from qiling.arch.models import MIPS_CPU_MODEL
from qiling.arch.register import QlRegisterManager
from qiling.const import QL_ARCH, QL_ENDIAN
@@ -20,8 +22,8 @@ class QlArchMIPS(QlArch):
type = QL_ARCH.MIPS
bits = 32
def __init__(self, ql: Qiling, endian: QL_ENDIAN):
super().__init__(ql)
def __init__(self, ql: Qiling, *, cputype: Optional[MIPS_CPU_MODEL], endian: QL_ENDIAN):
super().__init__(ql, cputype=cputype)
self._init_endian = endian
qiling/arch/models.py
+447
View File
@@ -0,0 +1,447 @@
#!/usr/bin/env python3
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
from enum import Enum
from typing import Union
class X86_CPU_MODEL(Enum):
# generic
GENERIC_QEMU64 = 0
GENERIC_KVM64 = 3
GENERIC_QEMU32 = 4
GENERIC_KVM32 = 5
# amd
AMD_PHENOM = 1
AMD_ATHLON = 11
AMD_OPTERON_G1 = 30
AMD_OPTERON_G2 = 31
AMD_OPTERON_G3 = 32
AMD_OPTERON_G4 = 33
AMD_OPTERON_G5 = 34
AMD_EPYC = 35
AMD_DHYANA = 36
AMD_EPYC_ROME = 37
# intel
INTEL_CORE2DUO = 2
INTEL_COREDUO = 6
INTEL_486 = 7
INTEL_PENTIUM = 8
INTEL_PENTIUM2 = 9
INTEL_PENTIUM3 = 10
INTEL_N270 = 12
INTEL_CONROE = 13
INTEL_PENRYN = 14
INTEL_NEHALEM = 15
INTEL_WESTMERE = 16
INTEL_SANDYBRIDGE = 17
INTEL_IVYBRIDGE = 18
INTEL_HASWELL = 19
INTEL_BROADWELL = 20
INTEL_SKYLAKE_CLIENT = 21
INTEL_SKYLAKE_SERVER = 22
INTEL_CASCADELAKE_SERVER = 23
INTEL_COOPERLAKE = 24
INTEL_ICELAKE_CLIENT = 25
INTEL_ICELAKE_SERVER = 26
INTEL_DENVERTON = 27
INTEL_SNOWRIDGE = 28
INTEL_KNIGHTSMILL = 29
class ARM_CPU_MODEL(Enum):
ARM_926 = 0
ARM_946 = 1
ARM_1026 = 2
ARM_1136_R2 = 3
ARM_1136 = 4
ARM_1176 = 5
ARM_11MPCORE = 6
ARM_CORTEX_M0 = 7
ARM_CORTEX_M3 = 8
ARM_CORTEX_M4 = 9
ARM_CORTEX_M7 = 10
ARM_CORTEX_M33 = 11
ARM_CORTEX_R5 = 12
ARM_CORTEX_R5F = 13
ARM_CORTEX_A7 = 14
ARM_CORTEX_A8 = 15
ARM_CORTEX_A9 = 16
ARM_CORTEX_A15 = 17
ARM_TI925T = 18
ARM_SA1100 = 19
ARM_SA1110 = 20
ARM_PXA250 = 21
ARM_PXA255 = 22
ARM_PXA260 = 23
ARM_PXA261 = 24
ARM_PXA262 = 25
ARM_PXA270 = 26
ARM_PXA270A0 = 27
ARM_PXA270A1 = 28
ARM_PXA270B0 = 29
ARM_PXA270B1 = 30
ARM_PXA270C0 = 31
ARM_PXA270C5 = 32
ARM_MAX = 33
class ARM64_CPU_MODEL(Enum):
ARM64_A57 = 0
ARM64_A53 = 1
ARM64_A72 = 2
ARM64_MAX = 3
class MIPS_CPU_MODEL(Enum):
MIPS32_4KC = 0
MIPS32_4KM = 1
MIPS32_4KECR1 = 2
MIPS32_4KEMR1 = 3
MIPS32_4KEC = 4
MIPS32_4KEM = 5
MIPS32_24KC = 6
MIPS32_24KEC = 7
MIPS32_24KF = 8
MIPS32_34KF = 9
MIPS32_74KF = 10
MIPS32_M14K = 11
MIPS32_M14KC = 12
MIPS32_P5600 = 13
MIPS32_MIPS32R6_GENERIC = 14
MIPS32_I7200 = 15
class PPC_CPU_MODEL(Enum):
PPC32_401 = 0
PPC32_401A1 = 1
PPC32_401B2 = 2
PPC32_401C2 = 3
PPC32_401D2 = 4
PPC32_401E2 = 5
PPC32_401F2 = 6
PPC32_401G2 = 7
PPC32_IOP480 = 8
PPC32_COBRA = 9
PPC32_403GA = 10
PPC32_403GB = 11
PPC32_403GC = 12
PPC32_403GCX = 13
PPC32_405D2 = 14
PPC32_405D4 = 15
PPC32_405CRA = 16
PPC32_405CRB = 17
PPC32_405CRC = 18
PPC32_405EP = 19
PPC32_405EZ = 20
PPC32_405GPA = 21
PPC32_405GPB = 22
PPC32_405GPC = 23
PPC32_405GPD = 24
PPC32_405GPR = 25
PPC32_405LP = 26
PPC32_NPE405H = 27
PPC32_NPE405H2 = 28
PPC32_NPE405L = 29
PPC32_NPE4GS3 = 30
PPC32_STB03 = 31
PPC32_STB04 = 32
PPC32_STB25 = 33
PPC32_X2VP4 = 34
PPC32_X2VP20 = 35
PPC32_440_XILINX = 36
PPC32_440_XILINX_W_DFPU = 37
PPC32_440EPA = 38
PPC32_440EPB = 39
PPC32_440EPX = 40
PPC32_460EXB = 41
PPC32_G2 = 42
PPC32_G2H4 = 43
PPC32_G2GP = 44
PPC32_G2LS = 45
PPC32_G2HIP3 = 46
PPC32_G2HIP4 = 47
PPC32_MPC603 = 48
PPC32_G2LE = 49
PPC32_G2LEGP = 50
PPC32_G2LELS = 51
PPC32_G2LEGP1 = 52
PPC32_G2LEGP3 = 53
PPC32_MPC5200_V10 = 54
PPC32_MPC5200_V11 = 55
PPC32_MPC5200_V12 = 56
PPC32_MPC5200B_V20 = 57
PPC32_MPC5200B_V21 = 58
PPC32_E200Z5 = 59
PPC32_E200Z6 = 60
PPC32_E300C1 = 61
PPC32_E300C2 = 62
PPC32_E300C3 = 63
PPC32_E300C4 = 64
PPC32_MPC8343 = 65
PPC32_MPC8343A = 66
PPC32_MPC8343E = 67
PPC32_MPC8343EA = 68
PPC32_MPC8347T = 69
PPC32_MPC8347P = 70
PPC32_MPC8347AT = 71
PPC32_MPC8347AP = 72
PPC32_MPC8347ET = 73
PPC32_MPC8347EP = 74
PPC32_MPC8347EAT = 75
PPC32_MPC8347EAP = 76
PPC32_MPC8349 = 77
PPC32_MPC8349A = 78
PPC32_MPC8349E = 79
PPC32_MPC8349EA = 80
PPC32_MPC8377 = 81
PPC32_MPC8377E = 82
PPC32_MPC8378 = 83
PPC32_MPC8378E = 84
PPC32_MPC8379 = 85
PPC32_MPC8379E = 86
PPC32_E500_V10 = 87
PPC32_E500_V20 = 88
PPC32_E500V2_V10 = 89
PPC32_E500V2_V20 = 90
PPC32_E500V2_V21 = 91
PPC32_E500V2_V22 = 92
PPC32_E500V2_V30 = 93
PPC32_E500MC = 94
PPC32_MPC8533_V10 = 95
PPC32_MPC8533_V11 = 96
PPC32_MPC8533E_V10 = 97
PPC32_MPC8533E_V11 = 98
PPC32_MPC8540_V10 = 99
PPC32_MPC8540_V20 = 100
PPC32_MPC8540_V21 = 101
PPC32_MPC8541_V10 = 102
PPC32_MPC8541_V11 = 103
PPC32_MPC8541E_V10 = 104
PPC32_MPC8541E_V11 = 105
PPC32_MPC8543_V10 = 106
PPC32_MPC8543_V11 = 107
PPC32_MPC8543_V20 = 108
PPC32_MPC8543_V21 = 109
PPC32_MPC8543E_V10 = 110
PPC32_MPC8543E_V11 = 111
PPC32_MPC8543E_V20 = 112
PPC32_MPC8543E_V21 = 113
PPC32_MPC8544_V10 = 114
PPC32_MPC8544_V11 = 115
PPC32_MPC8544E_V10 = 116
PPC32_MPC8544E_V11 = 117
PPC32_MPC8545_V20 = 118
PPC32_MPC8545_V21 = 119
PPC32_MPC8545E_V20 = 120
PPC32_MPC8545E_V21 = 121
PPC32_MPC8547E_V20 = 122
PPC32_MPC8547E_V21 = 123
PPC32_MPC8548_V10 = 124
PPC32_MPC8548_V11 = 125
PPC32_MPC8548_V20 = 126
PPC32_MPC8548_V21 = 127
PPC32_MPC8548E_V10 = 128
PPC32_MPC8548E_V11 = 129
PPC32_MPC8548E_V20 = 130
PPC32_MPC8548E_V21 = 131
PPC32_MPC8555_V10 = 132
PPC32_MPC8555_V11 = 133
PPC32_MPC8555E_V10 = 134
PPC32_MPC8555E_V11 = 135
PPC32_MPC8560_V10 = 136
PPC32_MPC8560_V20 = 137
PPC32_MPC8560_V21 = 138
PPC32_MPC8567 = 139
PPC32_MPC8567E = 140
PPC32_MPC8568 = 141
PPC32_MPC8568E = 142
PPC32_MPC8572 = 143
PPC32_MPC8572E = 144
PPC32_E600 = 145
PPC32_MPC8610 = 146
PPC32_MPC8641 = 147
PPC32_MPC8641D = 148
PPC32_601_V0 = 149
PPC32_601_V1 = 150
PPC32_601_V2 = 151
PPC32_602 = 152
PPC32_603 = 153
PPC32_603E_V1_1 = 154
PPC32_603E_V1_2 = 155
PPC32_603E_V1_3 = 156
PPC32_603E_V1_4 = 157
PPC32_603E_V2_2 = 158
PPC32_603E_V3 = 159
PPC32_603E_V4 = 160
PPC32_603E_V4_1 = 161
PPC32_603E7 = 162
PPC32_603E7T = 163
PPC32_603E7V = 164
PPC32_603E7V1 = 165
PPC32_603E7V2 = 166
PPC32_603P = 167
PPC32_604 = 168
PPC32_604E_V1_0 = 169
PPC32_604E_V2_2 = 170
PPC32_604E_V2_4 = 171
PPC32_604R = 172
PPC32_740_V1_0 = 173
PPC32_750_V1_0 = 174
PPC32_740_V2_0 = 175
PPC32_750_V2_0 = 176
PPC32_740_V2_1 = 177
PPC32_750_V2_1 = 178
PPC32_740_V2_2 = 179
PPC32_750_V2_2 = 180
PPC32_740_V3_0 = 181
PPC32_750_V3_0 = 182
PPC32_740_V3_1 = 183
PPC32_750_V3_1 = 184
PPC32_740E = 185
PPC32_750E = 186
PPC32_740P = 187
PPC32_750P = 188
PPC32_750CL_V1_0 = 189
PPC32_750CL_V2_0 = 190
PPC32_750CX_V1_0 = 191
PPC32_750CX_V2_0 = 192
PPC32_750CX_V2_1 = 193
PPC32_750CX_V2_2 = 194
PPC32_750CXE_V2_1 = 195
PPC32_750CXE_V2_2 = 196
PPC32_750CXE_V2_3 = 197
PPC32_750CXE_V2_4 = 198
PPC32_750CXE_V2_4B = 199
PPC32_750CXE_V3_0 = 200
PPC32_750CXE_V3_1 = 201
PPC32_750CXE_V3_1B = 202
PPC32_750CXR = 203
PPC32_750FL = 204
PPC32_750FX_V1_0 = 205
PPC32_750FX_V2_0 = 206
PPC32_750FX_V2_1 = 207
PPC32_750FX_V2_2 = 208
PPC32_750FX_V2_3 = 209
PPC32_750GL = 210
PPC32_750GX_V1_0 = 211
PPC32_750GX_V1_1 = 212
PPC32_750GX_V1_2 = 213
PPC32_750L_V2_0 = 214
PPC32_750L_V2_1 = 215
PPC32_750L_V2_2 = 216
PPC32_750L_V3_0 = 217
PPC32_750L_V3_2 = 218
PPC32_745_V1_0 = 219
PPC32_755_V1_0 = 220
PPC32_745_V1_1 = 221
PPC32_755_V1_1 = 222
PPC32_745_V2_0 = 223
PPC32_755_V2_0 = 224
PPC32_745_V2_1 = 225
PPC32_755_V2_1 = 226
PPC32_745_V2_2 = 227
PPC32_755_V2_2 = 228
PPC32_745_V2_3 = 229
PPC32_755_V2_3 = 230
PPC32_745_V2_4 = 231
PPC32_755_V2_4 = 232
PPC32_745_V2_5 = 233
PPC32_755_V2_5 = 234
PPC32_745_V2_6 = 235
PPC32_755_V2_6 = 236
PPC32_745_V2_7 = 237
PPC32_755_V2_7 = 238
PPC32_745_V2_8 = 239
PPC32_755_V2_8 = 240
PPC32_7400_V1_0 = 241
PPC32_7400_V1_1 = 242
PPC32_7400_V2_0 = 243
PPC32_7400_V2_1 = 244
PPC32_7400_V2_2 = 245
PPC32_7400_V2_6 = 246
PPC32_7400_V2_7 = 247
PPC32_7400_V2_8 = 248
PPC32_7400_V2_9 = 249
PPC32_7410_V1_0 = 250
PPC32_7410_V1_1 = 251
PPC32_7410_V1_2 = 252
PPC32_7410_V1_3 = 253
PPC32_7410_V1_4 = 254
PPC32_7448_V1_0 = 255
PPC32_7448_V1_1 = 256
PPC32_7448_V2_0 = 257
PPC32_7448_V2_1 = 258
PPC32_7450_V1_0 = 259
PPC32_7450_V1_1 = 260
PPC32_7450_V1_2 = 261
PPC32_7450_V2_0 = 262
PPC32_7450_V2_1 = 263
PPC32_7441_V2_1 = 264
PPC32_7441_V2_3 = 265
PPC32_7451_V2_3 = 266
PPC32_7441_V2_10 = 267
PPC32_7451_V2_10 = 268
PPC32_7445_V1_0 = 269
PPC32_7455_V1_0 = 270
PPC32_7445_V2_1 = 271
PPC32_7455_V2_1 = 272
PPC32_7445_V3_2 = 273
PPC32_7455_V3_2 = 274
PPC32_7445_V3_3 = 275
PPC32_7455_V3_3 = 276
PPC32_7445_V3_4 = 277
PPC32_7455_V3_4 = 278
PPC32_7447_V1_0 = 279
PPC32_7457_V1_0 = 280
PPC32_7447_V1_1 = 281
PPC32_7457_V1_1 = 282
PPC32_7457_V1_2 = 283
PPC32_7447A_V1_0 = 284
PPC32_7457A_V1_0 = 285
PPC32_7447A_V1_1 = 286
PPC32_7457A_V1_1 = 287
PPC32_7447A_V1_2 = 288
PPC32_7457A_V1_2 = 289
class RISCV_CPU_MODEL(Enum):
RISCV32_ANY = 0
RISCV32_BASE32 = 1
RISCV32_SIFIVE_E31 = 2
RISCV32_SIFIVE_U34 = 3
class RISCV64_CPU_MODEL(Enum):
RISCV64_ANY = 0
RISCV64_BASE64 = 1
RISCV64_SIFIVE_E51 = 2
RISCV64_SIFIVE_U54 = 3
QL_CPU = Union[
X86_CPU_MODEL,
ARM_CPU_MODEL,
ARM64_CPU_MODEL,
MIPS_CPU_MODEL,
PPC_CPU_MODEL,
RISCV_CPU_MODEL,
RISCV64_CPU_MODEL
]
__all__ = [
'X86_CPU_MODEL',
'ARM_CPU_MODEL',
'ARM64_CPU_MODEL',
'MIPS_CPU_MODEL',
'PPC_CPU_MODEL',
'RISCV_CPU_MODEL',
'RISCV64_CPU_MODEL',
'QL_CPU'
]
qiling/arch/ppc.py
+12 -1
View File
@@ -4,13 +4,16 @@
#
from functools import cached_property
from typing import Optional
from unicorn import Uc, UC_ARCH_PPC, UC_MODE_PPC32, UC_MODE_BIG_ENDIAN
from capstone import Cs, CS_ARCH_PPC, CS_MODE_32, CS_MODE_BIG_ENDIAN
from keystone import Ks, KS_ARCH_PPC, KS_MODE_PPC32, KS_MODE_BIG_ENDIAN
from qiling import Qiling
from qiling.arch.arch import QlArch
from qiling.arch import ppc_const
from qiling.arch.models import PPC_CPU_MODEL
from qiling.arch.register import QlRegisterManager
from qiling.const import QL_ARCH, QL_ENDIAN
@@ -19,9 +22,17 @@ class QlArchPPC(QlArch):
type = QL_ARCH.PPC
bits = 32
def __init__(self, ql: Qiling, *, cputype: Optional[PPC_CPU_MODEL] = None):
super().__init__(ql, cputype=cputype)
@cached_property
def uc(self) -> Uc:
return Uc(UC_ARCH_PPC, UC_MODE_PPC32 + UC_MODE_BIG_ENDIAN)
obj = Uc(UC_ARCH_PPC, UC_MODE_PPC32 + UC_MODE_BIG_ENDIAN)
if self.cpu is not None:
obj.ctl_set_cpu_model(self.cpu.value)
return obj
@cached_property
def regs(self) -> QlRegisterManager:
qiling/arch/riscv.py
+13 -2
View File
@@ -4,12 +4,15 @@
#
from functools import cached_property
from typing import Optional
from unicorn import Uc, UC_ARCH_RISCV, UC_MODE_RISCV32
from capstone import Cs
from keystone import Ks
from qiling import Qiling
from qiling.arch.arch import QlArch
from qiling.arch.models import RISCV_CPU_MODEL
from qiling.arch.register import QlRegisterManager
from qiling.arch import riscv_const
from qiling.arch.riscv_const import *
@@ -21,9 +24,17 @@ class QlArchRISCV(QlArch):
type = QL_ARCH.RISCV
bits = 32
def __init__(self, ql: Qiling, *, cputype: Optional[RISCV_CPU_MODEL] = None):
super().__init__(ql, cputype=cputype)
@cached_property
def uc(self) -> Uc:
return Uc(UC_ARCH_RISCV, UC_MODE_RISCV32)
obj = Uc(UC_ARCH_RISCV, UC_MODE_RISCV32)
if self.cpu is not None:
obj.ctl_set_cpu_model(self.cpu.value)
return obj
@cached_property
def regs(self) -> QlRegisterManager:
@@ -61,7 +72,7 @@ class QlArchRISCV(QlArch):
def init_context(self):
self.regs.pc = 0x08000000
def unicorn_exception_handler(self, ql, intno):
def unicorn_exception_handler(self, ql: Qiling, intno: int):
if intno == 2:
ql.log.warning(f'[{hex(self.regs.arch_pc)}] Illegal instruction')
qiling/arch/riscv64.py
+12 -1
View File
@@ -4,11 +4,14 @@
#
from functools import cached_property
from typing import Optional
from unicorn import Uc, UC_ARCH_RISCV, UC_MODE_RISCV64
from capstone import Cs
from keystone import Ks
from qiling import Qiling
from qiling.arch.models import RISCV64_CPU_MODEL
from qiling.arch.riscv_const import *
from qiling.const import QL_ARCH
from qiling.exception import QlErrorNotImplemented
@@ -20,9 +23,17 @@ class QlArchRISCV64(QlArchRISCV):
type = QL_ARCH.RISCV64
bits = 64
def __init__(self, ql: Qiling, *, cputype: Optional[RISCV64_CPU_MODEL] = None):
super().__init__(ql, cputype=cputype)
@cached_property
def uc(self) -> Uc:
return Uc(UC_ARCH_RISCV, UC_MODE_RISCV64)
obj = Uc(UC_ARCH_RISCV, UC_MODE_RISCV64)
if self.cpu is not None:
obj.ctl_set_cpu_model(self.cpu.value)
return obj
@cached_property
def disassembler(self) -> Cs:
qiling/arch/utils.py
+15 -2
View File
@@ -28,9 +28,22 @@ class QlArchUtils:
@lru_cache(maxsize=64)
def get_base_and_name(self, addr: int) -> Tuple[int, str]:
for begin, end, _, name, _ in self.ql.mem.map_info:
# executable images may be mapped in multiple consecutive regions, so locating
# an address within a region doesn't mean its base address is the base of the
# image. here we iterate through memory map regions as if they have been coalesced
# by label to find the image base address.
base_label = '?'
base_addr = -1
for begin, end, _, label, _ in self.ql.mem.map_info:
# reached a different image?
if label != base_label:
base_addr = begin
base_label = label
if begin <= addr < end:
return begin, basename(name)
return base_addr, basename(label)
return addr, '-'
qiling/arch/x86.py
+21 -12
View File
@@ -4,19 +4,40 @@
#
from functools import cached_property
from typing import Optional
from unicorn import Uc, UC_ARCH_X86, UC_MODE_16, UC_MODE_32, UC_MODE_64
from capstone import Cs, CS_ARCH_X86, CS_MODE_16, CS_MODE_32, CS_MODE_64
from keystone import Ks, KS_ARCH_X86, KS_MODE_16, KS_MODE_32, KS_MODE_64
from qiling import Qiling
from qiling.arch.arch import QlArch
from qiling.arch.msr import QlMsrManager
from qiling.arch.models import X86_CPU_MODEL
from qiling.arch.register import QlRegisterManager
from qiling.arch import x86_const
from qiling.const import QL_ARCH, QL_ENDIAN
class QlArchIntel(QlArch):
def __init__(self, ql: Qiling, *, cputype: Optional[X86_CPU_MODEL] = None):
super().__init__(ql, cputype=cputype)
@cached_property
def uc(self) -> Uc:
mode = {
16: UC_MODE_16,
32: UC_MODE_32,
64: UC_MODE_64
}[self.bits]
obj = Uc(UC_ARCH_X86, mode)
if self.cpu is not None:
obj.ctl_set_cpu_model(self.cpu.value)
return obj
@property
def endian(self) -> QL_ENDIAN:
return QL_ENDIAN.EL
@@ -33,10 +54,6 @@ class QlArchA8086(QlArchIntel):
type = QL_ARCH.A8086
bits = 16
@cached_property
def uc(self) -> Uc:
return Uc(UC_ARCH_X86, UC_MODE_16)
@cached_property
def regs(self) -> QlRegisterManager:
regs_map = dict(
@@ -63,10 +80,6 @@ class QlArchX86(QlArchIntel):
type = QL_ARCH.X86
bits = 32
@cached_property
def uc(self) -> Uc:
return Uc(UC_ARCH_X86, UC_MODE_32)
@cached_property
def regs(self) -> QlRegisterManager:
regs_map = dict(
@@ -97,10 +110,6 @@ class QlArchX8664(QlArchIntel):
type = QL_ARCH.X8664
bits = 64
@cached_property
def uc(self) -> Uc:
return Uc(UC_ARCH_X86, UC_MODE_64)
@cached_property
def regs(self) -> QlRegisterManager:
regs_map = dict(
qiling/const.py
+2 -1
View File
@@ -91,10 +91,11 @@ def __casefold_enum(e: Type[T]) -> Mapping[str, T]:
return dict((k.casefold(), v) for k, v in e.__members__.items())
debugger_map = __casefold_enum(QL_DEBUGGER)
endian_map = __casefold_enum(QL_ENDIAN)
arch_map = __casefold_enum(QL_ARCH)
os_map = __casefold_enum(QL_OS)
verbose_map = __casefold_enum(QL_VERBOSE)
debugger_map = __casefold_enum(QL_DEBUGGER)
arch_os_map = {
QL_ARCH.EVM : QL_OS.EVM,
qiling/core.py
+3 -1
View File
@@ -20,6 +20,7 @@ if TYPE_CHECKING:
from .hw.hw import QlHwManager
from .loader.loader import QlLoader
from .arch.models import QL_CPU
from .const import QL_ARCH, QL_ENDIAN, QL_OS, QL_STATE, QL_STOP, QL_VERBOSE, QL_ARCH_INTERPRETER, QL_OS_BAREMETAL
from .exception import QlErrorFileNotFound, QlErrorArch, QlErrorOsType
from .host import QlHost
@@ -38,6 +39,7 @@ class Qiling(QlCoreHooks, QlCoreStructs):
code: Optional[bytes] = None,
ostype: Optional[QL_OS] = None,
archtype: Optional[QL_ARCH] = None,
cputype: Optional[QL_CPU] = None,
verbose: QL_VERBOSE = QL_VERBOSE.DEFAULT,
profile: Optional[Union[str, Mapping]] = None,
console: bool = True,
@@ -147,7 +149,7 @@ class Qiling(QlCoreHooks, QlCoreStructs):
if endian is None:
endian = QL_ENDIAN.EL
self._arch = select_arch(archtype, endian, thumb)(self)
self._arch = select_arch(archtype, cputype, endian, thumb)(self)
# Once we finish setting up arch, we can init QlCoreStructs and QlCoreHooks
if not self.interpreter:
qiling/extensions/coverage/formats/base.py
+9 -6
View File
@@ -1,10 +1,12 @@
#!/usr/bin/env python3
#
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
from abc import ABC, abstractmethod
from qiling import Qiling
class QlBaseCoverage(ABC):
"""
@@ -12,14 +14,16 @@ class QlBaseCoverage(ABC):
To add support for a new coverage format, just derive from this class and implement
all the methods marked with the @abstractmethod decorator.
"""
def __init__(self):
def __init__(self, ql: Qiling):
super().__init__()
self.ql = ql
@property
@staticmethod
@abstractmethod
def FORMAT_NAME():
def FORMAT_NAME() -> str:
raise NotImplementedError
@abstractmethod
@@ -31,6 +35,5 @@ class QlBaseCoverage(ABC):
pass
@abstractmethod
def dump_coverage(self, coverage_file):
def dump_coverage(self, coverage_file: str):
pass
qiling/extensions/coverage/formats/drcov.py
+7 -6
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
#
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
@@ -17,6 +17,7 @@ class bb_entry(Structure):
("mod_id", c_uint16)
]
class QlDrCoverage(QlBaseCoverage):
"""
Collects emulated code coverage and formats it in accordance with the DynamoRIO based
@@ -29,12 +30,12 @@ class QlDrCoverage(QlBaseCoverage):
FORMAT_NAME = "drcov"
def __init__(self, ql):
super().__init__()
self.ql = ql
super().__init__(ql)
self.drcov_version = 2
self.drcov_flavor = 'drcov'
self.basic_blocks = []
self.bb_callback = None
self.drcov_flavor = 'drcov'
self.basic_blocks = []
self.bb_callback = None
@staticmethod
def block_callback(ql, address, size, self):
qiling/extensions/coverage/formats/history.py
+22 -22
View File
@@ -1,5 +1,5 @@
from __future__ import annotations
from typing import List, Tuple, TYPE_CHECKING, Union, Optional, Any
from typing import List, Sequence, Tuple, TYPE_CHECKING, Union, Optional, Any
if TYPE_CHECKING:
from qiling import Qiling
@@ -24,8 +24,8 @@ class History:
self.track_block_coverage()
def clear_history(self) -> None:
"""Clears the current state of the history
"""Clears the current state of the history
"""
self.history.clear()
@@ -33,7 +33,7 @@ class History:
"""Clears the current history hook from the Qiling instance
Returns:
None
None
"""
self.ql.hook_del(self.history_hook_handle)
@@ -56,13 +56,13 @@ class History:
try:
self.history.append(next(self.md.disasm(ins_bytes, address)))
except StopIteration:
# if this ever happens, then the unicorn/qiling is going to crash because it tried to execute
# if this ever happens, then the unicorn/qiling is going to crash because it tried to execute
# an instruction that it cant, so we are just not going to do anything
pass
def track_block_coverage(self) -> None:
"""Configures the history plugin to track all of the basic blocks that are executed. Removes any existing hooks
Returns:
None
"""
@@ -73,7 +73,7 @@ class History:
def track_instruction_coverage(self) -> None:
"""Configures the history plugin to track all of the instructions that are executed. Removes any existing hooks
Returns:
None
"""
@@ -82,9 +82,9 @@ class History:
self.history_hook_handle = self.ql.hook_code(self.__hook_block)
def get_ins_only_lib(self, libs: List[str]) -> List[CsInsn]:
def get_ins_only_lib(self, libs: Union[str, Sequence[str]]) -> List[CsInsn]:
"""Returns a list of addresses that have been executed that are only in mmaps for objects that match the regex of items in the list
Args:
libs (List[str]): A list of regex strings to match against the library names in the memory maps
@@ -93,18 +93,18 @@ class History:
Examples:
>>> history.get_ins_only_lib([".*libc.so.*", ".*libpthread.so.*"])
"""
"""
executable_maps = self.get_regex_matching_exec_maps(libs)
return [x for x in self.history if any(start <= x.address <= end for start, end, _, _, _ in executable_maps)]
def get_ins_exclude_lib(self, libs: List[str]) -> List[CsInsn]:
def get_ins_exclude_lib(self, libs: Union[str, Sequence[str]]) -> List[CsInsn]:
'''Returns a list of history instructions that are not in the libraries that match the regex in the libs list
Args:
libs (List[str]): A list of regex strings to match against the library names in the memory maps
Returns:
List[capstone.CsInsn]: A list of CsInsn that have been executed and are not in the memory maps that match the regex
@@ -115,7 +115,7 @@ class History:
executable_maps = self.get_regex_matching_exec_maps(libs)
return [h for h in self.history if not any(start <= h.address <= end for start, end, _, _, _ in executable_maps)]
def get_mem_map_from_addr(self, ins: Union[int, CsInsn]) -> Optional[Tuple]:
def get_mem_map_from_addr(self, ins: Union[int, CsInsn]) -> Optional[Tuple[int, int, str, str, str]]:
'''Returns the memory map that contains the instruction
Args:
@@ -125,7 +125,7 @@ class History:
Optional[Tuple]: A tuple that contains the memory map that contains the instruction
this tuple is in the format of (start_addr, end_addr, perms, name, path)
Examples:
Examples:
>>> history.get_mem_map_from_addr(0x7ffff7dd1b97)
'''
@@ -134,12 +134,12 @@ class History:
assert isinstance(ins, int)
return next((x for x in self.ql.mem.get_mapinfo() if x[0] <= ins and x[1] >= ins), None)
return next((x for x in self.ql.mem.get_mapinfo() if x[0] <= ins <= x[1]), None)
def get_regex_matching_exec_maps(self, libs: List[str]) -> List[Tuple]:
def get_regex_matching_exec_maps(self, libs: Union[str, Sequence[str]]) -> List[Tuple[int, int, str, str, str]]:
'''Returns a list of tuples for current mmaps whose names match the regex of libs in the list
This is a wrapper around ql.mem.get_mapinfo() and just filters the results by the regex of the library names
This is a wrapper around ql.mem.get_mapinfo() and just filters the results by the regex of the library names
and also only returns maps that are executable
Args:
@@ -158,9 +158,9 @@ class History:
libs = [libs]
# filter the history list by the library name, using a list of regex
regex = [re.compile(lib) for lib in libs]
regex = [re.compile(lib) for lib in libs]
# filter the list of tuples
# filter the list of tuples
# so that we return only the ones where the library name matches the regex
regex_matching_libs = [x for x in self.ql.mem.get_mapinfo() if any(r.match(x[3]) for r in regex)]
qiling/extensions/coverage/utils.py
+21 -8
View File
@@ -1,15 +1,25 @@
#!/usr/bin/env python3
#
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
from __future__ import annotations
from contextlib import contextmanager
from typing import Dict, TYPE_CHECKING, List, Type, TypeVar
from .formats import *
from .formats.base import QlBaseCoverage
if TYPE_CHECKING:
from qiling import Qiling
CT = TypeVar('CT', bound=QlBaseCoverage)
# Returns subclasses recursively.
def get_all_subclasses(cls):
def get_all_subclasses(cls: Type[CT]) -> List[Type[CT]]:
all_subclasses = []
for subclass in cls.__subclasses__():
@@ -18,21 +28,24 @@ def get_all_subclasses(cls):
return all_subclasses
class CoverageFactory():
class CoverageFactory:
def __init__(self):
self.coverage_collectors = {subcls.FORMAT_NAME:subcls for subcls in get_all_subclasses(base.QlBaseCoverage)}
self.coverage_collectors: Dict[str, Type[QlBaseCoverage]] = {subcls.FORMAT_NAME: subcls for subcls in get_all_subclasses(QlBaseCoverage)}
@property
def formats(self):
return self.coverage_collectors.keys()
def get_coverage_collector(self, ql, name):
def get_coverage_collector(self, ql: Qiling, name: str) -> QlBaseCoverage:
return self.coverage_collectors[name](ql)
factory = CoverageFactory()
@contextmanager
def collect_coverage(ql, name, coverage_file):
def collect_coverage(ql: Qiling, name: str, coverage_file: str):
"""
Context manager for emulating a given piece of code with coverage collection turned on.
Example:
@@ -42,9 +55,9 @@ def collect_coverage(ql, name, coverage_file):
cov = factory.get_coverage_collector(ql, name)
cov.activate()
try:
yield
finally:
cov.deactivate()
if coverage_file:
cov.dump_coverage(coverage_file)
cov.dump_coverage(coverage_file)
qiling/extensions/multitask.py
+49 -32
View File
@@ -1,6 +1,5 @@
# Lazymio ([email protected])
from typing import Dict, List
from unicorn import *
from unicorn.x86_const import UC_X86_REG_EIP, UC_X86_REG_RIP
from unicorn.arm64_const import UC_ARM64_REG_PC
@@ -16,6 +15,9 @@ import gevent.threadpool
import gevent.lock
import threading
from typing import Dict, List, Optional
# This class is named UnicornTask be design since it's not a
# real thread. The expected usage is to inherit this class
# and overwrite specific methods.
@@ -23,10 +25,10 @@ import threading
# This class is a friend class of MultiTaskUnicorn
class UnicornTask:
def __init__(self, uc: Uc, begin: int, end: int, task_id = None):
def __init__(self, uc: Uc, begin: int, end: int, task_id=None):
self._uc = uc
self._begin = begin
self._end = end
self._end = end
self._stop_request = False
self._ctx = None
self._task_id = None
@@ -34,46 +36,55 @@ class UnicornTask:
self._mode = self._uc._mode
@property
def pc(self):
def pc(self) -> int:
""" Get current PC of the thread. This property should only be accessed when
the task is running.
"""
raw_pc = self._raw_pc()
if (self._uc.reg_read(UC_ARM_REG_CPSR) & (1 << 5)):
return raw_pc | 1
else:
return raw_pc
def _raw_pc(self):
if self._arch == UC_ARCH_ARM:
return raw_pc | ((self._uc.reg_read(UC_ARM_REG_CPSR) >> 5) & 0b1)
return raw_pc
def _raw_pc(self) -> int:
# This extension is designed to be independent of Qiling, so let's
# do this manually...
pc_reg = 0 # invalid reg
if self._arch == UC_ARCH_X86:
if (self._mode & UC_MODE_32) != 0:
return self._uc.reg_read(UC_X86_REG_EIP)
elif (self._mode & UC_MODE_64) != 0:
return self._uc.reg_read(UC_X86_REG_RIP)
if self._mode & UC_MODE_32:
pc_reg = UC_X86_REG_EIP
elif self._mode & UC_MODE_64:
pc_reg = UC_X86_REG_RIP
elif self._arch == UC_ARCH_MIPS:
return self._uc.reg_read(UC_MIPS_REG_PC)
pc_reg = UC_MIPS_REG_PC
elif self._arch == UC_ARCH_ARM:
return self._uc.reg_read(UC_ARM_REG_PC)
pc_reg = UC_ARM_REG_PC
elif self._arch == UC_ARCH_ARM64:
return self._uc.reg_read(UC_ARM64_REG_PC)
elif self._arch == UC_ARCH_PPC:
return self._uc.reg_read(UC_PPC_REG_PC)
elif self._arch == UC_ARCH_M68K:
return self._uc.reg_read(UC_M68K_REG_PC)
elif self._arch == UC_ARCH_SPARC:
return self._uc.reg_read(UC_SPARC_REG_PC)
elif self._arch == UC_ARCH_RISCV:
return self._uc.reg_read(UC_RISCV_REG_PC)
pc_reg = UC_ARM64_REG_PC
# Really?
return 0
elif self._arch == UC_ARCH_PPC:
pc_reg = UC_PPC_REG_PC
elif self._arch == UC_ARCH_M68K:
pc_reg = UC_M68K_REG_PC
elif self._arch == UC_ARCH_SPARC:
pc_reg = UC_SPARC_REG_PC
elif self._arch == UC_ARCH_RISCV:
pc_reg = UC_RISCV_REG_PC
return self._uc.reg_read(pc_reg) if pc_reg else 0
def _reach_end(self):
# We may stop due to the scheduler asks us to, so check it manually.
#print(f"{hex(self._raw_pc())} {hex(self._end)}")
# print(f"{hex(self._raw_pc())} {hex(self._end)}")
return self._raw_pc() == self._end
def save(self):
@@ -106,13 +117,14 @@ class UnicornTask:
"""
pass
# This manages nested uc_emu_start calls and is designed as a friend
# class of MultiTaskUnicorn.
class NestedCounter:
def __init__(self, mtuc: "MultiTaskUnicorn"):
self._mtuc = mtuc
def __enter__(self, *args, **kwargs):
self._mtuc._nested_started += 1
return self
@@ -120,6 +132,7 @@ class NestedCounter:
def __exit__(self, *args, **kwargs):
self._mtuc._nested_started -= 1
# This mimic a Unicorn object by maintaining the same interface.
# If no task is registered, the behavior is exactly the same as
# a normal unicorn.
@@ -137,14 +150,18 @@ class NestedCounter:
# the same time.
class MultiTaskUnicorn(Uc):
def __init__(self, arch, mode, interval: int = 100):
def __init__(self, arch: int, mode: int, cpu: Optional[int], interval: Optional[int] = 100):
""" Create a MultiTaskUnicorn object.
Interval: Sceduling interval in **ms**. The longger interval, the better
performance but less interrupts.
"""
super().__init__(arch, mode)
if cpu is not None:
self.ctl_set_cpu_model(cpu)
self._interval = interval
self._tasks = {} # type: Dict[int, UnicornTask]
self._tasks = {} # type: Dict[int, UnicornTask]
self._task_id_counter = 2000
self._to_stop = False
self._cur_utk_id = None
@@ -271,7 +288,7 @@ class MultiTaskUnicorn(Uc):
def task_exit(self, utk_id):
""" Stop a task.
utk_id: The id returned from task_create.
"""
if utk_id not in self._tasks:
@@ -369,4 +386,4 @@ class MultiTaskUnicorn(Uc):
gevent.joinall(workset, raise_error=True)
if len(self._tasks) == 0:
self._multitask_enabled = False
self._multitask_enabled = False
qiling/loader/dos.py
+17 -11
View File
@@ -1,15 +1,17 @@
#!/usr/bin/env python3
#
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
import sys, traceback
import sys
import traceback
from .loader import QlLoader
from qiling import Qiling
from qiling.os.disk import QlDisk
# @see: http://pinvoke.net/default.aspx/Structures.IMAGE_DOS_HEADER
class ComParser:
'''Most basic COM file parser.
@@ -20,17 +22,21 @@ class ComParser:
nbytes = ql.unpack16(data[2:4]) or 0x200 # number of bytes in last block; 0 means it is fully populated
nblocks = ql.unpack16(data[4:6]) # number of blocks used
hdrpgs = ql.unpack16(data[8:10]) # number of paragraphs taken up by the header
self.size = (nblocks - 1) * 0x200 + nbytes
self.header_size = hdrpgs * 0x10
self.init_ss = ql.unpack16(data[14:16])
self.init_sp = ql.unpack16(data[16:18])
self.init_ip = ql.unpack16(data[20:22])
self.init_cs = ql.unpack16(data[22:24])
class QlLoaderDOS(QlLoader):
def __init__(self, ql: Qiling):
super(QlLoaderDOS, self).__init__(ql)
self.ql = ql
super().__init__(ql)
self.old_excepthook = sys.excepthook
# Hack to print all exceptions if curses has been setup.
@@ -38,6 +44,7 @@ class QlLoaderDOS(QlLoader):
if self.ql.os.stdscr is not None:
tbmsg = "".join(traceback.format_exception(tp, value, tb))
self.ql.log.info(f"{tbmsg}")
self.old_excepthook(tp, value, tb)
def run(self):
@@ -49,9 +56,9 @@ class QlLoaderDOS(QlLoader):
with open(path, "rb") as f:
content = f.read()
cs = int(profile.get("COM", "start_cs"), 0)
ip = int(profile.get("COM", "start_ip"), 0)
sp = int(profile.get("COM", "start_sp"), 0)
cs = profile.getint("COM", "start_cs")
ip = profile.getint("COM", "start_ip")
sp = profile.getint("COM", "start_sp")
ss = cs
base_address = (cs << 4) + ip
@@ -69,7 +76,7 @@ class QlLoaderDOS(QlLoader):
ss = com.init_ss
base_address = 0
content = content[0x80:]
content = content[com.header_size:]
elif path.endswith(".DOS_MBR"):
with open(path, "rb") as f:
@@ -100,8 +107,7 @@ class QlLoaderDOS(QlLoader):
self.stack_address = (ss << 4) + sp
self.start_address = (cs << 4) + ip
self.stack_size = int(profile.get("COM", "stack_size"), 0)
self.ticks_per_second = profile.getfloat("KERNEL", "ticks_per_second")
self.stack_size = profile.getint("COM", "stack_size")
# map the entire system memory
self.ql.mem.map(0, 0x100000, info="[FULL]")
@@ -110,4 +116,4 @@ class QlLoaderDOS(QlLoader):
self.load_address = base_address
self.ql.os.entry_point = self.start_address
sys.excepthook = self.excepthook
sys.excepthook = self.excepthook
qiling/loader/elf.py
+17 -6
View File
@@ -7,7 +7,7 @@ import io
import os
from enum import IntEnum
from typing import Optional, Sequence, Mapping, Tuple
from typing import AnyStr, Optional, Sequence, Mapping, Tuple
from elftools.common.utils import preserve_stream_pos
from elftools.elf.constants import P_FLAGS, SH_FLAGS
@@ -143,7 +143,7 @@ class QlLoaderELF(QlLoader):
return prot
def load_with_ld(self, elffile: ELFFile, stack_addr: int, load_address: int, argv: Sequence[str] = [], env: Mapping[str, str] = {}):
def load_with_ld(self, elffile: ELFFile, stack_addr: int, load_address: int, argv: Sequence[str] = [], env: Mapping[AnyStr, AnyStr] = {}):
def load_elf_segments(elffile: ELFFile, load_address: int, info: str):
# get list of loadable segments; these segments will be loaded to memory
@@ -272,17 +272,23 @@ class QlLoaderELF(QlLoader):
elf_table = bytearray()
new_stack = stack_addr
def __push_str(top: int, s: str) -> int:
"""Write a string to stack memory and adjust the top of stack accordingly.
def __push_bytes(top: int, b: bytes) -> int:
"""Write bytes to stack memory and adjust the top of stack accordingly.
Top of stack remains aligned to pointer size
"""
data = s.encode('latin') + b'\x00'
data = b + b'\x00'
top = self.ql.mem.align(top - len(data), self.ql.arch.pointersize)
self.ql.mem.write(top, data)
return top
def __push_str(top: int, s: str) -> int:
"""A convinient method for writing a string to stack memory.
"""
return __push_bytes(top, s.encode('latin'))
# write argc
elf_table.extend(self.ql.pack(len(argv)))
@@ -296,7 +302,12 @@ class QlLoaderELF(QlLoader):
# write env
for k, v in env.items():
new_stack = __push_str(new_stack, f'{k}={v}')
_k = k if isinstance(k, bytes) else k.encode('latin')
_v = v if isinstance(v, bytes) else v.encode('latin')
pair = b'='.join((_k, _v))
new_stack = __push_bytes(new_stack, pair)
elf_table.extend(self.ql.pack(new_stack))
# add a nullptr sentinel
qiling/log.py
+6 -6
View File
@@ -198,13 +198,13 @@ def setup_logger(ql: Qiling, log_file: Optional[str], console: bool, log_overrid
handler.setFormatter(formatter)
log.addHandler(handler)
log.setLevel(logging.INFO)
# optimize logging speed by avoiding the collection of unnecesary logging properties
logging._srcfile = None
logging.logThreads = False
logging.logProcesses = False
logging.logMultiprocessing = False
# optimize logging speed by avoiding the collection of unnecesary logging properties
logging._srcfile = None
logging.logThreads = False
logging.logProcesses = False
logging.logMultiprocessing = False
log.setLevel(logging.INFO)
return log
qiling/os/dos/dos.py
+21 -19
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
#
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
@@ -15,26 +15,27 @@ from qiling.os.os import QlOs
from .interrupts import handlers
# @see: https://en.wikipedia.org/wiki/FLAGS_register
class Flags(IntEnum):
CF = (1 << 0) # carry
PF = (1 << 2) # parity
AF = (1 << 4) # alignment
ZF = (1 << 6) # zero
SF = (1 << 7) # sign
TF = (1 << 8) # trap
IF = (1 << 9) # interrupt
DF = (1 << 10) # direction
OF = (1 << 11) # overflow
IOPL = (3 << 12) # io privilege
CF = (1 << 0) # carry
PF = (1 << 2) # parity
AF = (1 << 4) # alignment
ZF = (1 << 6) # zero
SF = (1 << 7) # sign
TF = (1 << 8) # trap
IF = (1 << 9) # interrupt
DF = (1 << 10) # direction
OF = (1 << 11) # overflow
IOPL = (3 << 12) # io privilege
class QlOsDos(QlOs):
type = QL_OS.DOS
def __init__(self, ql: Qiling):
super(QlOsDos, self).__init__(ql)
super().__init__(ql)
self.ql = ql
self.hook_syscall()
# used by int 21h
@@ -46,7 +47,11 @@ class QlOsDos(QlOs):
self.revese_color_pairs = {}
self.stdscr = None
self.dos_ver = int(self.ql.profile.get("KERNEL", "version"), 0)
kconf = self.ql.profile["KERNEL"]
self.dos_ver = kconf.getint("version")
self.ticks_per_second = kconf.getfloat("ticks_per_second")
def __del__(self):
# resume terminal
@@ -83,7 +88,7 @@ class QlOsDos(QlOs):
func = self.user_defined_api[QL_INTERCEPT.CALL].get(intinfo) or handlers.get(intno)
onenter = self.user_defined_api[QL_INTERCEPT.ENTER].get(intinfo)
onexit = self.user_defined_api[QL_INTERCEPT.EXIT].get(intinfo)
onexit = self.user_defined_api[QL_INTERCEPT.EXIT].get(intinfo)
if onenter is not None:
onenter(ql)
@@ -103,14 +108,11 @@ class QlOsDos(QlOs):
if self.ql.exit_point is not None:
self.exit_point = self.ql.exit_point
if self.ql.entry_point is not None:
self.ql.loader.elf_entry = self.ql.entry_point
else:
if self.ql.entry_point is None:
self.ql.entry_point = self.ql.loader.start_address
if not self.ql.code:
self.start_time = datetime.now()
self.ticks_per_second = self.ql.loader.ticks_per_second
try:
self.ql.emu_start(self.ql.entry_point, self.exit_point, self.ql.timeout, self.ql.count)
qiling/os/dos/interrupts/int13.py
+38 -33
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
#
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
@@ -10,40 +10,44 @@ from qiling import Qiling
from .. import utils
class DiskError(IntEnum):
NoError = 0
BadCommand = 1
AddressNotFound = 2
DiskWriteProtectError = 3
SectorNotFound = 4
FixedDiskResetFailed = 5
DiskChangedOrRemoved = 6
BadFixedDiskParameterTable = 7
DMAOverrun = 8
DMAAcessAcrossBoundary = 9
BadFixedDiskSectorFlag = 10
BadFixedDiskCylinder = 11
UnsupportedTrack = 12
InvalidNumberofSectors = 13
FixedDiskControlledDataAdressDetected = 14
FixedDiskDMAArbitrationLevelOutofRange = 15
ECCErrorOnRead = 16
RecoverableFixedDiskDataError = 17
ControllerError = 32
SeekFailure = 64
Timeout = 128
FixedDiskDriveNotReady = 170
FixedDiskUndefinedError = 187
FixedDiskWriteFault = 204
FixedDiskStatusError = 224
SenseOperationFailed = 255
NoError = 0
BadCommand = 1
AddressNotFound = 2
DiskWriteProtectError = 3
SectorNotFound = 4
FixedDiskResetFailed = 5
DiskChangedOrRemoved = 6
BadFixedDiskParameterTable = 7
DMAOverrun = 8
DMAAcessAcrossBoundary = 9
BadFixedDiskSectorFlag = 10
BadFixedDiskCylinder = 11
UnsupportedTrack = 12
InvalidNumberofSectors = 13
FixedDiskControlledDataAdressDetected = 14
FixedDiskDMAArbitrationLevelOutofRange = 15
ECCErrorOnRead = 16
RecoverableFixedDiskDataError = 17
ControllerError = 32
SeekFailure = 64
Timeout = 128
FixedDiskDriveNotReady = 170
FixedDiskUndefinedError = 187
FixedDiskWriteFault = 204
FixedDiskStatusError = 224
SenseOperationFailed = 255
def parse_dap(dapbs):
return struct.unpack("<BBHHHQ", dapbs)
def __leaf_00(ql: Qiling):
ql.os.clear_cf()
def __leaf_02(ql: Qiling):
idx = ql.arch.regs.dl
@@ -142,17 +146,18 @@ def __leaf_43(ql: Qiling):
ql.os.clear_cf()
ql.arch.regs.ah = 0
# @see: https://en.wikipedia.org/wiki/INT_13H
def handler(ql: Qiling):
ah = ql.arch.regs.ah
leaffunc = {
0x00 : __leaf_00,
0x02 : __leaf_02,
0x08 : __leaf_08,
0x41 : __leaf_41,
0x42 : __leaf_42,
0x43 : __leaf_43
0x00: __leaf_00,
0x02: __leaf_02,
0x08: __leaf_08,
0x41: __leaf_41,
0x42: __leaf_42,
0x43: __leaf_43
}.get(ah)
if leaffunc is None:
qiling/os/dos/interrupts/int1a.py
+23 -13
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
#
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
@@ -9,6 +9,7 @@ from qiling import Qiling
from .. import utils
def __set_elapsed_ticks(ql: Qiling):
now = datetime.now()
ticks = int((now - ql.os.start_time).total_seconds() * ql.os.ticks_per_second)
@@ -16,14 +17,17 @@ def __set_elapsed_ticks(ql: Qiling):
ql.arch.regs.cx = (ticks >> 16) & 0xffff
ql.arch.regs.dx = (ticks >> 0) & 0xffff
def __leaf_00(ql: Qiling):
__set_elapsed_ticks(ql)
ql.arch.regs.al = 0
def __leaf_01(ql: Qiling):
__set_elapsed_ticks(ql)
def __leaf_02_03(ql: Qiling):
now = datetime.now()
@@ -34,6 +38,7 @@ def __leaf_02_03(ql: Qiling):
ql.os.clear_cf()
def __leaf_04_05(ql: Qiling):
now = datetime.now()
@@ -45,37 +50,42 @@ def __leaf_04_05(ql: Qiling):
ql.os.clear_cf()
def __leaf_06_07_09(ql: Qiling):
# TODO: Implement clock interrupt.
ql.os.set_cf()
def __leaf_08(ql: Qiling):
pass
def __leaf_0a(ql: Qiling):
now = datetime.now()
ql.arch.regs.cx = (now - datetime(1980, 1, 1)).days
def __leaf_0b(ql: Qiling):
pass
def handler(ql: Qiling):
ah = ql.arch.regs.ah
leaffunc = {
0x00 : __leaf_00,
0x01 : __leaf_01,
0x02 : __leaf_02_03,
0x03 : __leaf_02_03,
0x04 : __leaf_04_05,
0x05 : __leaf_04_05,
0x06 : __leaf_06_07_09,
0x07 : __leaf_06_07_09,
0x08 : __leaf_08,
0x09 : __leaf_06_07_09,
0x0a : __leaf_0a,
0x0b : __leaf_0b
0x00: __leaf_00,
0x01: __leaf_01,
0x02: __leaf_02_03,
0x03: __leaf_02_03,
0x04: __leaf_04_05,
0x05: __leaf_04_05,
0x06: __leaf_06_07_09,
0x07: __leaf_06_07_09,
0x08: __leaf_08,
0x09: __leaf_06_07_09,
0x0a: __leaf_0a,
0x0b: __leaf_0b
}.get(ah)
if leaffunc is None:
qiling/os/dos/interrupts/int21.py
+18 -18
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
#
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
@@ -135,23 +135,23 @@ def handler(ql: Qiling):
ah = ql.arch.regs.ah
leaffunc = {
0x02 : __leaf_02,
0x06 : __leaf_02,
0x09 : __leaf_09,
0x0c : __leaf_0c,
0x25 : __leaf_25,
0x26 : __leaf_26,
0x30 : __leaf_30,
0x33 : __leaf_33,
0x35 : __leaf_35,
0x3c : __leaf_3c,
0x3d : __leaf_3d,
0x3e : __leaf_3e,
0x3f : __leaf_3f,
0x40 : __leaf_40,
0x41 : __leaf_41,
0x43 : __leaf_43,
0x4c : __leaf_4c
0x02: __leaf_02,
0x06: __leaf_02,
0x09: __leaf_09,
0x0c: __leaf_0c,
0x25: __leaf_25,
0x26: __leaf_26,
0x30: __leaf_30,
0x33: __leaf_33,
0x35: __leaf_35,
0x3c: __leaf_3c,
0x3d: __leaf_3d,
0x3e: __leaf_3e,
0x3f: __leaf_3f,
0x40: __leaf_40,
0x41: __leaf_41,
0x43: __leaf_43,
0x4c: __leaf_4c
}.get(ah)
if leaffunc is None:
qiling/os/macos/macos.py
+8 -2
View File
@@ -18,14 +18,20 @@ from qiling.os.macos.events.macos import QlMacOSEvManager
from qiling.os.macos.events.macos_policy import QlMacOSPolicy
from qiling.os.macos.events.macos_structs import mac_policy_list_t
from qiling.os.macos.structs import kmod_info_t, POINTER64
from qiling.os.posix.syscall.abi import arm
class QlOsMacos(QlOsPosix):
type = QL_OS.MACOS
def __init__(self, ql: Qiling):
super(QlOsMacos, self).__init__(ql)
super().__init__(ql)
# NOTE: it appears that MacOS only supports Intel arch, but we put this
# here anyway for completion
if ql.arch.type is QL_ARCH.ARM64:
self.syscall_abi = arm.QlAArch64MacOS(ql.arch)
self.ql = ql
self.fcall = QlFunctionCall(ql, intel.macosx64(ql.arch))
self.ql.counter = 0
qiling/os/memory.py
+14 -16
View File
@@ -26,7 +26,7 @@ class QlMemoryManager:
https://github.com/zeropointdynamics/zelos/blob/master/src/zelos/memory.py
"""
def __init__(self, ql: Qiling):
def __init__(self, ql: Qiling, pagesize: int = 0x1000):
self.ql = ql
self.map_info: List[MapInfoEntry] = []
self.mmio_cbs = {}
@@ -40,13 +40,10 @@ class QlMemoryManager:
if ql.arch.bits not in bit_stuff:
raise QlErrorStructConversion("Unsupported Qiling architecture for memory manager")
max_addr = bit_stuff[ql.arch.bits]
self.max_addr = max_addr
self.max_mem_addr = max_addr
self.max_mem_addr = bit_stuff[ql.arch.bits]
# memory page size
self.pagesize = 0x1000
self.pagesize = pagesize
# make sure pagesize is a power of 2
assert self.pagesize & (self.pagesize - 1) == 0, 'pagesize has to be a power of 2'
@@ -134,7 +131,7 @@ class QlMemoryManager:
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 = None
info_idx: int = -1
for idx, map_info in enumerate(self.map_info):
if mem_s >= map_info[0] and mem_e <= map_info[1]:
@@ -377,7 +374,7 @@ class QlMemoryManager:
self.write(addr, __pack(value))
def search(self, needle: Union[bytes, Pattern[bytes]], begin: Optional[int] = None, end: Optional[int] = None) -> Sequence[int]:
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.
Args:
@@ -398,7 +395,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 size effects
# 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)]
results = []
@@ -477,7 +474,6 @@ class QlMemoryManager:
yield (p_lbound, p_ubound)
def is_available(self, addr: int, size: int) -> bool:
"""Query whether the memory range starting at `addr` and is of length of `size` bytes
is available for allocation.
@@ -527,7 +523,7 @@ class QlMemoryManager:
# memory space bounds (exclusive)
mem_lbound = 0
mem_ubound = self.max_addr + 1
mem_ubound = self.max_mem_addr + 1
if minaddr is None:
minaddr = mem_lbound
@@ -587,8 +583,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, mem_p = perms)
self.change_mapinfo(aligned_address, aligned_address + aligned_size, perms)
def map(self, addr: int, size: int, perms: int = UC_PROT_ALL, info: Optional[str] = None):
"""Map a new memory range.
@@ -640,14 +635,18 @@ class QlMemoryManager:
self.mmio_cbs[(addr, addr + size)] = (read_cb, write_cb)
# A Simple Heap Implementation
class Chunk:
def __init__(self, address: int, size: int):
self.inuse = True
self.address = address
self.size = size
class QlMemoryHeap:
"""A Simple Heap Implementation.
"""
def __init__(self, ql: Qiling, start_address: int, end_address: int):
self.ql = ql
self.chunks: List[Chunk] = []
@@ -758,8 +757,7 @@ class QlMemoryHeap:
# clear all memory regions alloc
def clear(self):
for chunk in self.chunks:
chunk.inuse = False
self.chunks.clear()
for addr, size in self.mem_alloc:
self.ql.mem.unmap(addr, size)
qiling/os/posix/const_mapping.py
+8 -4
View File
@@ -37,15 +37,19 @@ def _flags_mapping(value: int, flags_map: Mapping[str, int]) -> str:
def ql_open_flag_mapping(ql: Qiling, flags: int) -> int:
def flag_mapping(flags, mapping_name, mapping_from, mapping_to, host_os, virt_os):
def flag_mapping(flags, mapping_name, mapping_from, mapping_to, host_os):
ret = 0
for n in mapping_name:
if mapping_from[n] is None or mapping_to[n] is None:
continue
if (flags & mapping_from[n]) == mapping_from[n]:
ret = ret | mapping_to[n]
if (host_os == QL_OS.WINDOWS and virt_os != QL_OS.WINDOWS):
ret = ret | mapping_to['O_BINARY']
if host_os is QL_OS.WINDOWS:
ret = ret | mapping_to['O_BINARY']
return ret
f = {}
@@ -89,7 +93,7 @@ def ql_open_flag_mapping(ql: Qiling, flags: int) -> int:
if f == t:
return flags
return flag_mapping(flags, open_flags_name, f, t, host_os, virt_os)
return flag_mapping(flags, open_flags_name, f, t, host_os)
def mmap_flag_mapping(flags):
qiling/os/posix/msq.py
+56
View File
@@ -0,0 +1,56 @@
#!/usr/bin/env python3
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
from typing import Dict, Tuple, Optional
from collections import deque
from qiling.os.posix.const import MSGMNB
class QlMsgBuf:
def __init__(self, mtype: int, mtext: bytes) -> None:
self.mtype = mtype
self.mtext = mtext
# vaguely reflects a msqid64_ds structure
class QlMsqId:
def __init__(self, key: int, uid: int, gid: int, mode: int) -> None:
# ipc64_perm
self.key = key
self.uid = uid
self.gid = gid
self.mode = mode
self.queue = deque(maxlen=MSGMNB)
def __len__(self):
return len(self.queue)
class QlMsq:
def __init__(self) -> None:
self.__msq: Dict[int, QlMsqId] = {}
self.__id: int = 0x0F000000
def __len__(self) -> int:
return len(self.__msq)
def add(self, msq: QlMsqId) -> int:
msqid = self.__id
self.__msq[msqid] = msq
self.__id += 0x1000
return msqid
def remove(self, msqid: int) -> None:
del self.__msq[msqid]
def get_by_key(self, key: int) -> Tuple[int, Optional[QlMsqId]]:
return next(((msqid, msqobj) for msqid, msqobj in self.__msq.items() if msqobj.key == key), (-1, None))
def get_by_id(self, msqid: int) -> Optional[QlMsqId]:
return self.__msq.get(msqid, None)
qiling/os/posix/posix.py
+18 -205
View File
@@ -3,68 +3,21 @@
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
from collections import deque
from inspect import signature, Parameter
from typing import Dict, TextIO, Tuple, Union, Callable, IO, List, Optional
from unicorn.arm64_const import UC_ARM64_REG_X8, UC_ARM64_REG_X16
from unicorn.arm_const import (
UC_ARM_REG_R0, UC_ARM_REG_R1, UC_ARM_REG_R2, UC_ARM_REG_R3,
UC_ARM_REG_R4, UC_ARM_REG_R5, UC_ARM_REG_R7, UC_ARM_REG_R12
)
from unicorn.mips_const import UC_MIPS_REG_V0
from unicorn.x86_const import (
UC_X86_REG_EAX, UC_X86_REG_EBX, UC_X86_REG_ECX, UC_X86_REG_EDX,
UC_X86_REG_ESI, UC_X86_REG_EDI, UC_X86_REG_EBP, UC_X86_REG_RDI,
UC_X86_REG_RSI, UC_X86_REG_RDX, UC_X86_REG_R10, UC_X86_REG_R8,
UC_X86_REG_R9, UC_X86_REG_RAX
)
from unicorn.riscv_const import UC_RISCV_REG_A7
from unicorn.ppc_const import UC_PPC_REG_0
from typing import Dict, TextIO, Union, Callable, IO, List, Optional
from qiling import Qiling
from qiling.cc import QlCC, intel, arm, mips, riscv, ppc
from qiling.const import QL_ARCH, QL_OS, QL_INTERCEPT
from qiling.const import QL_ARCH, QL_INTERCEPT
from qiling.exception import QlErrorSyscallNotFound
from qiling.os.os import QlOs
from qiling.os.posix.const import MSGMNB, NR_OPEN, errors
from qiling.os.posix.const import NR_OPEN, errors
from qiling.os.posix.msq import QlMsq
from qiling.os.posix.shm import QlShm
from qiling.os.posix.syscall.abi import QlSyscallABI, arm, intel, mips, ppc, riscv
from qiling.utils import ql_get_module, ql_get_module_function
SYSCALL_PREF: str = f'ql_syscall_'
class intel32(intel.QlIntel32):
_argregs = (UC_X86_REG_EBX, UC_X86_REG_ECX, UC_X86_REG_EDX, UC_X86_REG_ESI, UC_X86_REG_EDI, UC_X86_REG_EBP)
class intel64(intel.QlIntel64):
_argregs = (UC_X86_REG_RDI, UC_X86_REG_RSI, UC_X86_REG_RDX, UC_X86_REG_R10, UC_X86_REG_R8, UC_X86_REG_R9)
class aarch32(arm.aarch32):
_argregs = (UC_ARM_REG_R0, UC_ARM_REG_R1, UC_ARM_REG_R2, UC_ARM_REG_R3, UC_ARM_REG_R4, UC_ARM_REG_R5)
class aarch64(arm.aarch64):
pass
class mipso32(mips.mipso32):
# TODO: should it be part of the standard mipso32 cc?
def setReturnValue(self, value: int):
if -1134 < value < 0:
a3return = 1
value = -value
else:
a3return = 0
self.arch.regs.v0 = value
self.arch.regs.a3 = a3return
class riscv32(riscv.riscv):
pass
class riscv64(riscv.riscv):
pass
class ppc(ppc.ppc):
pass
class QlFileDes:
def __init__(self):
@@ -92,104 +45,10 @@ class QlFileDes:
self.__fds = fds
# vaguely reflects a shmid64_ds structure
class QlShmId:
def __init__(self, key: int, uid: int, gid: int, mode: int, segsz: int) -> None:
# ipc64_perm
self.key = key
self.uid = uid
self.gid = gid
self.mode = mode
self.segsz = segsz
# track the memory locations this segment is currently attached to
self.attach: List[int] = []
class QlShm:
def __init__(self) -> None:
self.__shm: Dict[int, QlShmId] = {}
self.__id: int = 0x0F000000
def __len__(self) -> int:
return len(self.__shm)
def add(self, shm: QlShmId) -> int:
shmid = self.__id
self.__shm[shmid] = shm
self.__id += 0x1000
return shmid
def remove(self, shmid: int) -> None:
del self.__shm[shmid]
def get_by_key(self, key: int) -> Tuple[int, Optional[QlShmId]]:
return next(((shmid, shmobj) for shmid, shmobj in self.__shm.items() if shmobj.key == key), (-1, None))
def get_by_id(self, shmid: int) -> Optional[QlShmId]:
return self.__shm.get(shmid, None)
def get_by_attaddr(self, shmaddr: int) -> Optional[QlShmId]:
return next((shmobj for shmobj in self.__shm.values() if shmobj.attach.count(shmaddr) > 0), None)
class QlMsgBuf:
def __init__(self, mtype: int, mtext: bytes) -> None:
self.mtype = mtype
self.mtext = mtext
# vaguely reflects a msqid64_ds structure
class QlMsqId:
def __init__(self, key: int, uid: int, gid: int, mode: int) -> None:
# ipc64_perm
self.key = key
self.uid = uid
self.gid = gid
self.mode = mode
self.queue = deque(maxlen=MSGMNB)
def __len__(self):
return len(self.queue)
class QlMsq:
def __init__(self) -> None:
self.__msq: Dict[int, QlMsqId] = {}
self.__id: int = 0x0F000000
def __len__(self) -> int:
return len(self.__msq)
def add(self, msq: QlMsqId) -> int:
msqid = self.__id
self.__msq[msqid] = msq
self.__id += 0x1000
return msqid
def remove(self, msqid: int) -> None:
del self.__msq[msqid]
def get_by_key(self, key: int) -> Tuple[int, Optional[QlMsqId]]:
return next(((msqid, msqobj) for msqid, msqobj in self.__msq.items() if msqobj.key == key), (-1, None))
def get_by_id(self, msqid: int) -> Optional[QlMsqId]:
return self.__msq.get(msqid, None)
class QlOsPosix(QlOs):
def __init__(self, ql: Qiling):
super().__init__(ql)
self.ql = ql
self.sigaction_act = [0] * 256
conf = self.profile['KERNEL']
@@ -208,34 +67,15 @@ class QlOsPosix(QlOs):
QL_INTERCEPT.EXIT: {}
}
self.__syscall_id_reg = {
QL_ARCH.ARM64 : UC_ARM64_REG_X8,
QL_ARCH.ARM : UC_ARM_REG_R7,
QL_ARCH.MIPS : UC_MIPS_REG_V0,
QL_ARCH.X86 : UC_X86_REG_EAX,
QL_ARCH.X8664 : UC_X86_REG_RAX,
QL_ARCH.RISCV : UC_RISCV_REG_A7,
QL_ARCH.RISCV64 : UC_RISCV_REG_A7,
QL_ARCH.PPC : UC_PPC_REG_0
}[self.ql.arch.type]
# handle some special cases
if (self.ql.arch.type == QL_ARCH.ARM64) and (self.type == QL_OS.MACOS):
self.__syscall_id_reg = UC_ARM64_REG_X16
elif (self.ql.arch.type == QL_ARCH.ARM) and (self.type == QL_OS.QNX):
self.__syscall_id_reg = UC_ARM_REG_R12
# TODO: use abstract to access __syscall_cc and __syscall_id_reg by defining a system call class
self.__syscall_cc: QlCC = {
QL_ARCH.ARM64 : aarch64,
QL_ARCH.ARM : aarch32,
QL_ARCH.MIPS : mipso32,
QL_ARCH.X86 : intel32,
QL_ARCH.X8664 : intel64,
QL_ARCH.RISCV : riscv32,
QL_ARCH.RISCV64 : riscv64,
QL_ARCH.PPC : ppc
self.syscall_abi: QlSyscallABI = {
QL_ARCH.ARM64: arm.QlAArch64,
QL_ARCH.ARM: arm.QlAArch32,
QL_ARCH.MIPS: mips.QlMipsO32,
QL_ARCH.X86: intel.QlIntel32,
QL_ARCH.X8664: intel.QlIntel64,
QL_ARCH.RISCV: riscv.QlRiscV32,
QL_ARCH.RISCV64: riscv.QlRiscV64,
QL_ARCH.PPC: ppc.QlPPC
}[self.ql.arch.type](self.ql.arch)
# select syscall mapping function based on emulated OS and architecture
@@ -308,7 +148,6 @@ class QlOsPosix(QlOs):
else:
self.function_hook.add_function_hook(target, handler, intercept)
@staticmethod
def getNameFromErrorCode(ret: int) -> str:
"""Return the hex representation of a return value and if possible
@@ -325,7 +164,7 @@ class QlOsPosix(QlOs):
return f'{ret:#x}{f" ({errors[-ret]})" if -ret in errors else f""}'
def load_syscall(self):
syscall_id = self.get_syscall()
syscall_id = self.syscall_abi.get_id()
syscall_name = self.syscall_mapper(syscall_id)
# get syscall on-enter hook (if any)
@@ -360,7 +199,7 @@ class QlOsPosix(QlOs):
param_names = [info.name for info in param_names[1:] if info.kind == Parameter.POSITIONAL_OR_KEYWORD]
# read parameter values
params = [self.__syscall_cc.getRawParam(i) for i in range(len(param_names))]
params = self.syscall_abi.get_params(len(param_names))
try:
# if set, fire up the on-enter hook and let it override original args set
@@ -382,7 +221,7 @@ class QlOsPosix(QlOs):
# set return value
if retval is not None:
self.__syscall_cc.setReturnValue(retval)
self.syscall_abi.set_return_value(retval)
except KeyboardInterrupt:
raise
@@ -414,32 +253,6 @@ class QlOsPosix(QlOs):
if self.ql.debug_stop:
raise QlErrorSyscallNotFound(f'Syscall not found: {syscall_name}')
def get_syscall(self) -> int:
if self.ql.arch.type == QL_ARCH.ARM:
# support arm-oabi
# @see: https://marcin.juszkiewicz.com.pl/download/tables/syscalls.html
# @see: https://github.com/rootkiter/Reverse-bins/blob/master/syscall_header/armv4l_unistd.h
# @see: https://github.com/unicorn-engine/unicorn/issues/1137
# read the instruction we have just emulated
isize = 2 if self.ql.arch.is_thumb else self.ql.arch.pointersize
ibytes = self.ql.mem.read_ptr(self.ql.arch.regs.arch_pc - isize, isize)
# mask off the opcode, which is the most significant byte
svc_imm = ibytes & ((1 << ((isize - 1) * 8)) - 1)
# arm-oabi
if svc_imm >= 0x900000:
return svc_imm - 0x900000
if svc_imm > 0:
return svc_imm
return self.ql.arch.regs.read(self.__syscall_id_reg)
def set_syscall_return(self, retval: int):
self.__syscall_cc.setReturnValue(retval)
@property
def fd(self):
return self._fd
qiling/os/posix/shm.py
+51
View File
@@ -0,0 +1,51 @@
#!/usr/bin/env python3
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
from typing import Dict, List, Tuple, Optional
# vaguely reflects a shmid64_ds structure
class QlShmId:
def __init__(self, key: int, uid: int, gid: int, mode: int, segsz: int) -> None:
# ipc64_perm
self.key = key
self.uid = uid
self.gid = gid
self.mode = mode
self.segsz = segsz
# track the memory locations this segment is currently attached to
self.attach: List[int] = []
class QlShm:
def __init__(self) -> None:
self.__shm: Dict[int, QlShmId] = {}
self.__id: int = 0x0F000000
def __len__(self) -> int:
return len(self.__shm)
def add(self, shm: QlShmId) -> int:
shmid = self.__id
self.__shm[shmid] = shm
self.__id += 0x1000
return shmid
def remove(self, shmid: int) -> None:
del self.__shm[shmid]
def get_by_key(self, key: int) -> Tuple[int, Optional[QlShmId]]:
return next(((shmid, shmobj) for shmid, shmobj in self.__shm.items() if shmobj.key == key), (-1, None))
def get_by_id(self, shmid: int) -> Optional[QlShmId]:
return self.__shm.get(shmid, None)
def get_by_attaddr(self, shmaddr: int) -> Optional[QlShmId]:
return next((shmobj for shmobj in self.__shm.values() if shmobj.attach.count(shmaddr) > 0), None)
qiling/os/posix/syscall/abi/__init__.py
+60
View File
@@ -0,0 +1,60 @@
#!/usr/bin/env python3
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
from __future__ import annotations
from typing import TYPE_CHECKING, ClassVar, Sequence, Tuple
if TYPE_CHECKING:
from qiling.arch.arch import QlArch
class QlSyscallABI:
"""System call ABI common implementation.
"""
_idreg: ClassVar[int]
_argregs: ClassVar[Sequence[int]]
_retreg: ClassVar[int]
def __init__(self, arch: QlArch) -> None:
"""Initialize a system call ABI instance.
Args:
arch: underlying architecture instance
"""
self.arch = arch
def get_id(self) -> int:
"""Read system call ID number.
Returns: system call number
"""
return self.arch.regs.read(self._idreg) # type: ignore [uc funny annot]
def get_params(self, count: int) -> Tuple[int, ...]:
"""Read system call arguments.
Args:
count: number of arguments to read
Returns: a tuple containing system call arguments values
"""
if count > len(self._argregs):
raise ValueError(f'requested {count} arguments but only {len(self._argregs)} slots are defined')
return tuple(self.arch.regs.read(reg) for reg in self._argregs[:count]) # type: ignore [uc funny annot]
def set_return_value(self, value: int) -> None:
"""Set the system call return value.
Args:
value: a numeric value to set
"""
self.arch.regs.write(self._retreg, value)
qiling/os/posix/syscall/abi/arm.py
+67
View File
@@ -0,0 +1,67 @@
#!/usr/bin/env python3
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
from unicorn.arm64_const import (
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_X8, UC_ARM64_REG_X16
)
from unicorn.arm_const import (
UC_ARM_REG_R0, UC_ARM_REG_R1, UC_ARM_REG_R2, UC_ARM_REG_R3,
UC_ARM_REG_R4, UC_ARM_REG_R5, UC_ARM_REG_R7, UC_ARM_REG_R12
)
from qiling.os.posix.syscall.abi import QlSyscallABI
class QlAArch32(QlSyscallABI):
"""System call ABI for ARM-based systems.
"""
_idreg = UC_ARM_REG_R7
_argregs = (UC_ARM_REG_R0, UC_ARM_REG_R1, UC_ARM_REG_R2, UC_ARM_REG_R3, UC_ARM_REG_R4, UC_ARM_REG_R5)
_retreg = UC_ARM_REG_R0
def get_id(self) -> int:
# the syscall number of a svc / swi instruction needs to be manually extracted.
# here we read the instruction we have just emulated and extract the immediate
# number by masking off the the opcode
isize = 2 if self.arch.is_thumb else self.arch.pointersize
ibytes = self.arch.utils.ql.mem.read_ptr(self.arch.regs.arch_pc - isize, isize)
# mask off the opcode, which is the most significant byte
svc_imm = ibytes & ((1 << ((isize - 1) * 8)) - 1)
# arm-oabi
if svc_imm >= 0x900000:
return svc_imm - 0x900000
if svc_imm > 0:
return svc_imm
return super().get_id()
class QlAArch32QNX(QlAArch32):
"""QNX ABI override
"""
_idreg = UC_ARM_REG_R12
class QlAArch64(QlSyscallABI):
"""System call ABI for ARM64-based systems.
"""
_idreg = UC_ARM64_REG_X8
_argregs = (UC_ARM64_REG_X0, UC_ARM64_REG_X1, UC_ARM64_REG_X2, UC_ARM64_REG_X3, UC_ARM64_REG_X4, UC_ARM64_REG_X5)
_retreg = UC_ARM64_REG_X0
class QlAArch64MacOS(QlAArch64):
"""MacOS ABI override
"""
_idreg = UC_ARM64_REG_X16
qiling/os/posix/syscall/abi/intel.py
+30
View File
@@ -0,0 +1,30 @@
#!/usr/bin/env python3
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
from unicorn.x86_const import (
UC_X86_REG_EAX, UC_X86_REG_EBX, UC_X86_REG_ECX, UC_X86_REG_EDX,
UC_X86_REG_ESI, UC_X86_REG_EDI, UC_X86_REG_EBP, UC_X86_REG_RDI,
UC_X86_REG_RSI, UC_X86_REG_RDX, UC_X86_REG_R10, UC_X86_REG_R8,
UC_X86_REG_R9, UC_X86_REG_RAX
)
from qiling.os.posix.syscall.abi import QlSyscallABI
class QlIntel32(QlSyscallABI):
"""System call ABI for Intel-based 32-bit systems.
"""
_idreg = UC_X86_REG_EAX
_argregs = (UC_X86_REG_EBX, UC_X86_REG_ECX, UC_X86_REG_EDX, UC_X86_REG_ESI, UC_X86_REG_EDI, UC_X86_REG_EBP)
_retreg = UC_X86_REG_EAX
class QlIntel64(QlSyscallABI):
"""System call ABI for Intel-based 64-bit systems.
"""
_idreg = UC_X86_REG_RAX
_argregs = (UC_X86_REG_RDI, UC_X86_REG_RSI, UC_X86_REG_RDX, UC_X86_REG_R10, UC_X86_REG_R8, UC_X86_REG_R9)
_retreg = UC_X86_REG_RAX
qiling/os/posix/syscall/abi/mips.py
+41
View File
@@ -0,0 +1,41 @@
#!/usr/bin/env python3
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
from typing import Tuple
from unicorn.mips_const import UC_MIPS_REG_V0, UC_MIPS_REG_A0, UC_MIPS_REG_A1, UC_MIPS_REG_A2, UC_MIPS_REG_A3
from qiling.os.posix.syscall.abi import QlSyscallABI
class QlMipsO32(QlSyscallABI):
"""System call ABI for MIPS O32 systems.
See: https://www.linux-mips.org/wiki/Syscall
"""
_idreg = UC_MIPS_REG_V0
_argregs = (UC_MIPS_REG_A0, UC_MIPS_REG_A1, UC_MIPS_REG_A2, UC_MIPS_REG_A3)
_retreg = UC_MIPS_REG_V0
def __get_stack_params(self, count: int) -> Tuple[int, ...]:
"""Get system call parameters passed on stack.
"""
shadowed = len(self._argregs)
return tuple(self.arch.stack_read(self.arch.pointersize * (i + shadowed)) for i in range(count))
def get_params(self, count: int) -> Tuple[int, ...]:
num_reg_args = len(self._argregs)
stack_args = self.__get_stack_params(count - num_reg_args)
reg_args = super().get_params(min(count, num_reg_args))
return reg_args + stack_args
def set_return_value(self, value: int) -> None:
a3, v0 = (1, -value) if -1134 < value < 0 else (0, value)
self.arch.regs.v0 = v0
self.arch.regs.a3 = a3
qiling/os/posix/syscall/abi/ppc.py
+19
View File
@@ -0,0 +1,19 @@
#!/usr/bin/env python3
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
from unicorn.ppc_const import (
UC_PPC_REG_0, UC_PPC_REG_3, UC_PPC_REG_4, UC_PPC_REG_5,
UC_PPC_REG_6, UC_PPC_REG_7, UC_PPC_REG_8
)
from qiling.os.posix.syscall.abi import QlSyscallABI
class QlPPC(QlSyscallABI):
"""System call ABI for PowerPC systems.
"""
_idreg = UC_PPC_REG_0
_argregs = (UC_PPC_REG_3, UC_PPC_REG_4, UC_PPC_REG_5, UC_PPC_REG_6, UC_PPC_REG_7, UC_PPC_REG_8)
_retreg = UC_PPC_REG_3
qiling/os/posix/syscall/abi/riscv.py
+22
View File
@@ -0,0 +1,22 @@
#!/usr/bin/env python3
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
from unicorn.riscv_const import (
UC_RISCV_REG_A0, UC_RISCV_REG_A1, UC_RISCV_REG_A2, UC_RISCV_REG_A3,
UC_RISCV_REG_A4, UC_RISCV_REG_A5, UC_RISCV_REG_A7
)
from qiling.os.posix.syscall.abi import QlSyscallABI
class QlRiscV32(QlSyscallABI):
"""System call ABI for RISCV systems.
"""
_idreg = UC_RISCV_REG_A7
_argregs = (UC_RISCV_REG_A0, UC_RISCV_REG_A1, UC_RISCV_REG_A2, UC_RISCV_REG_A3, UC_RISCV_REG_A4, UC_RISCV_REG_A5)
_retreg = UC_RISCV_REG_A0
QlRiscV64 = QlRiscV32
qiling/os/posix/syscall/msg.py
+12 -9
View File
@@ -1,27 +1,24 @@
from typing import Optional
from qiling import Qiling
from qiling.os.posix.const import *
from qiling.os.posix.posix import QlMsqId, QlMsgBuf
from qiling.os.posix.msq import QlMsqId, QlMsgBuf
def __find_msg(msq: QlMsqId, msgtyp: int, msgflg: int) -> Optional[QlMsgBuf]:
# peek at a specific queue item
if msgflg & MSG_COPY:
if msgtyp >= len(msq.queue):
return -1 # ENOMSG
return msg.queue[msgtyp]
return msq.queue[msgtyp] if msgtyp < len(msq.queue) else None
if msgtyp == 0:
predicate = lambda msg: True
elif msgtype > 0:
elif msgtyp > 0:
if msgflg & MSG_EXCEPT:
predicate = lambda msg: msg.mtype != msgtyp
else:
predicate = lambda msg: msg.mtype == msgtyp
elif msgtype < 0:
elif msgtyp < 0:
predicate = lambda msg: msg.mtype <= -msgtyp
return next((msg for msg in msq.queue if predicate(msg)), None)
@@ -41,12 +38,13 @@ def __perms(ql: Qiling, msq: QlMsqId, flag: int) -> int:
if ql.os.uid == msq.uid:
granted_mode >>= 6
# is there some bit set in requested_mode but not in granted_mode?
# is there some bit set in requested_mode but not in granted_mode?
if request_mode & ~granted_mode & 0o007:
return -1 # EACCES
return 0
def ql_syscall_msgget(ql: Qiling, key: int, msgflg: int):
def __create_msq(key: int, flags: int) -> int:
"""Create a new message queue for the specified key.
@@ -92,6 +90,7 @@ def ql_syscall_msgget(ql: Qiling, key: int, msgflg: int):
return msqid
def ql_syscall_msgsnd(ql: Qiling, msqid: int, msgp: int, msgsz: int, msgflg: int):
msq = ql.os.msq.get_by_id(msqid)
@@ -122,7 +121,7 @@ def ql_syscall_msgrcv(ql: Qiling, msqid: int, msgp: int, msgsz: int, msgtyp: int
if msq is None:
return -1 # EINVAL
if msgflg & MSG_COPY:
if msgflg & MSG_EXCEPT or not (msgflg & IPC_NOWAIT):
return -1 # EINVAL
@@ -137,6 +136,9 @@ def ql_syscall_msgrcv(ql: Qiling, msqid: int, msgp: int, msgsz: int, msgtyp: int
if msg is not None:
break
if msgflg & MSG_COPY:
return -1 # ENOMSG
if msgflg & IPC_NOWAIT:
return -1 # ENOMSG
@@ -156,6 +158,7 @@ def ql_syscall_msgrcv(ql: Qiling, msqid: int, msgp: int, msgsz: int, msgtyp: int
return sz # Success
__all__ = [
'ql_syscall_msgget',
'ql_syscall_msgsnd',
qiling/os/posix/syscall/sched.py
+1 -1
View File
@@ -107,7 +107,7 @@ def ql_syscall_clone(ql: Qiling, flags: int, child_stack: int, parent_tidptr: in
ql.arch.regs.arch_pc += list(ql.arch.disassembler.disasm_lite(bytes(ql.mem.read(ql.arch.regs.arch_pc, 4)), ql.arch.regs.arch_pc))[0][1]
ql.log.debug(f"Fix pc for child thread to {hex(ql.arch.regs.arch_pc)}")
ql.os.set_syscall_return(0)
ql.os.syscall_abi.set_return_value(0)
th.save()
if th is None or f_th is None:
qiling/os/posix/syscall/shm.py
+1 -1
View File
@@ -9,7 +9,7 @@ from qiling import Qiling
from qiling.const import QL_ARCH
from qiling.exception import QlMemoryMappedError
from qiling.os.posix.const import *
from qiling.os.posix.posix import QlShmId
from qiling.os.posix.shm import QlShmId
def ql_syscall_shmget(ql: Qiling, key: int, size: int, shmflg: int):
qiling/os/posix/syscall/unistd.py
+1 -1
View File
@@ -213,7 +213,7 @@ def virtual_abspath_at(ql: Qiling, vpath: str, dirfd: int) -> Optional[str]:
# syscall params are read as unsigned int by default. until we fix that
# broadly, this is a workaround to turn fd into a signed value
dirfd = __as_signed(dirfd, ql.arch.bits)
dirfd = __as_signed(dirfd & ((1 << 32) - 1), 32)
# </WORKAROUND>
if dirfd == AT_FDCWD:
qiling/os/qnx/qnx.py
+4 -2
View File
@@ -8,6 +8,7 @@ from unicorn import UcError
from qiling import Qiling
from qiling.arch import arm_utils
from qiling.os.posix.posix import QlOsPosix
from qiling.os.posix.syscall.abi import arm as arm_abi
from qiling.os.qnx.const import NTO_SIDE_CHANNEL, SYSMGR_PID, SYSMGR_CHID, SYSMGR_COID
from qiling.os.qnx.helpers import QnxConn
from qiling.os.qnx.structs import _thread_local_storage
@@ -23,9 +24,10 @@ class QlOsQnx(QlOsPosix):
type = QL_OS.QNX
def __init__(self, ql: Qiling):
super(QlOsQnx, self).__init__(ql)
super().__init__(ql)
self.ql = ql
if ql.arch.type is QL_ARCH.ARM:
self.syscall_abi = arm_abi.QlAArch32QNX(ql.arch)
cc: QlCC = {
QL_ARCH.X86 : intel.cdecl,
qiling/utils.py
+16 -12
View File
@@ -20,9 +20,10 @@ from typing import TYPE_CHECKING, Any, Callable, Mapping, Optional, Tuple, TypeV
from unicorn import UC_ERR_READ_UNMAPPED, UC_ERR_FETCH_UNMAPPED
from qiling.exception import *
from qiling.arch.models import QL_CPU
from qiling.const import QL_ARCH, QL_ENDIAN, QL_OS, QL_DEBUGGER
from qiling.const import debugger_map, arch_map, os_map, arch_os_map
from qiling.exception import *
if TYPE_CHECKING:
from qiling import Qiling
@@ -293,11 +294,10 @@ def ql_guess_emu_env(path: str) -> Tuple[Optional[QL_ARCH], Optional[QL_OS], Opt
def select_loader(ostype: QL_OS, libcache: bool) -> QlClassInit['QlLoader']:
if ostype is QL_OS.WINDOWS:
kwargs = {'libcache': libcache}
kwargs = {}
else:
kwargs = {}
if ostype is QL_OS.WINDOWS:
kwargs['libcache'] = libcache
module = {
QL_OS.LINUX : r'elf',
@@ -370,17 +370,21 @@ def select_debugger(options: Union[str, bool]) -> Optional[QlClassInit['QlDebugg
return None
def select_arch(archtype: QL_ARCH, endian: QL_ENDIAN, thumb: bool) -> QlClassInit['QlArch']:
def select_arch(archtype: QL_ARCH, cputype: Optional[QL_CPU], endian: QL_ENDIAN, thumb: bool) -> QlClassInit['QlArch']:
kwargs = {}
# skip cpu model for evm
if archtype is not QL_ARCH.EVM:
kwargs['cputype'] = cputype
# set endianess and thumb mode for arm-based archs
if archtype is QL_ARCH.ARM:
kwargs = {'endian': endian, 'thumb': thumb}
kwargs['endian'] = endian
kwargs['thumb'] = thumb
# set endianess for mips arch
elif archtype is QL_ARCH.MIPS:
kwargs = {'endian': endian}
else:
kwargs = {}
kwargs['endian'] = endian
module = {
QL_ARCH.A8086 : r'x86',
@@ -430,7 +434,7 @@ def profile_setup(ostype: QL_OS, user_config: Optional[Union[str, dict]]):
int_converter = partial(int, base=0)
config = ConfigParser(converters={'int': int_converter})
qiling_home = Path(inspect.getfile(inspect.currentframe())).parent
qiling_home = Path(inspect.getfile(profile_setup)).parent
os_profile = qiling_home / 'profiles' / f'{ostype.name.lower()}.ql'
# read default profile first
qltool
+101 -80
View File
@@ -10,7 +10,7 @@ import ast
import pickle
from pprint import pprint
from typing import TYPE_CHECKING, Mapping, Type
from typing import TYPE_CHECKING, Any, AnyStr, Dict, Literal, Mapping, Type, overload
from unicorn import __version__ as uc_ver
from qiling import __version__ as ql_ver
@@ -18,7 +18,7 @@ from qiling import __version__ as ql_ver
from qiling import Qiling
from qiling.arch import utils as arch_utils
from qiling.debugger.qdb import QlQdb
from qiling.const import QL_VERBOSE, QL_ENDIAN, os_map, arch_map, verbose_map
from qiling.const import QL_ENDIAN, QL_VERBOSE, endian_map, os_map, arch_map, verbose_map
from qiling.extensions.coverage import utils as cov_utils
from qiling.extensions import report
@@ -27,9 +27,19 @@ if TYPE_CHECKING:
from enum import Enum
# read code from file
def read_file(fname: str):
with open(fname, "rb") as f:
@overload
def read_file(fname: AnyStr) -> str: ...
@overload
def read_file(fname: AnyStr, mode: Literal['b']) -> bytes: ...
def read_file(fname: AnyStr, mode: Literal['b', ''] = ''):
"""Read file contents.
"""
with open(fname, f'r{mode}') as f:
content = f.read()
return content
@@ -59,66 +69,64 @@ def __make_enum_arg(enum_rmap: Mapping[str, 'Enum'], aliases: Mapping[str, str]
return __enum_arg
__arg_endian = __make_enum_arg(endian_map)
__arg_archtype = __make_enum_arg(arch_map, {'x86_64': 'x8664', 'riscv32': 'riscv'})
__arg_ostype = __make_enum_arg(os_map, {'darwin': 'macos'})
__arg_verbose = __make_enum_arg(verbose_map)
def handle_code(options: argparse.Namespace):
archendian = {
'little': QL_ENDIAN.EL,
'big' : QL_ENDIAN.EB
}[options.endian]
def handle_code(options: argparse.Namespace) -> Dict[str, Any]:
if options.format == 'hex':
if options.input is not None:
print("Load HEX from ARGV")
code = str(options.input).strip("\\\\x").split("x")
code = "".join(code).strip()
code = bytes.fromhex(code)
print("Loading HEX from ARGV")
payload = str(options.input)
elif options.filename is not None:
print("Load HEX from FILE")
code = str(read_file(options.filename)).strip('b\'').strip('\\n')
code = code.strip('x').split("\\\\x")
code = "".join(code).strip()
code = bytes.fromhex(code)
print("Loading HEX from FILE")
payload = read_file(options.filename)
else:
print("ERROR: File not found")
print("Error: please specify an input source for hex")
exit(1)
elif options.format == 'asm':
print("Load ASM from FILE")
assembly = read_file(options.filename)
code = bytes.fromhex(payload.replace(r'\x', ''))
assembler = arch_utils.assembler(options.arch, archendian, options.thumb)
code, _ = assembler.asm(assembly)
elif options.format == 'asm':
print("Loading ASM from FILE")
if options.filename is None:
print("Error: please specify an input source for asm")
exit(1)
payload = read_file(options.filename)
# assemble the payload and turn it into opcodes bytes
assembler = arch_utils.assembler(options.arch, options.endian, options.thumb)
code, _ = assembler.asm(payload)
code = bytes(code)
elif options.format == 'bin':
print("Load BIN from FILE")
if options.filename is not None:
code = read_file(options.filename)
else:
print("ERROR: File not found")
print("Loading BIN from FILE")
if options.filename is None:
print("Error: please specify an input source for bin")
exit(1)
ql = Qiling(
rootfs=options.rootfs,
env=options.env,
code=code,
ostype=options.os,
archtype=options.arch,
verbose=options.verbose,
profile=options.profile,
filter=options.filter,
endian=archendian,
thumb=options.thumb,
)
code = read_file(options.filename, 'b')
return ql
ql_args = {
'rootfs': options.rootfs,
'code': code,
'ostype': options.os,
'archtype': options.arch,
'endian': options.endian,
'thumb': options.thumb
}
return ql_args
def handle_run(options: argparse.Namespace):
def handle_run(options: argparse.Namespace) -> Dict[str, Any]:
effective_argv = []
# with argv
@@ -132,26 +140,12 @@ def handle_run(options: argparse.Namespace):
else:
print("ERROR: Command error!")
ql = Qiling(
argv=effective_argv,
rootfs=options.rootfs,
env=options.env,
verbose=options.verbose,
profile=options.profile,
console=options.console,
log_file=options.log_file,
log_plain=options.log_plain,
multithread=options.multithread,
filter=options.filter,
libcache=options.libcache
)
ql_args = {
'argv': effective_argv,
'rootfs': options.rootfs
}
# attach Qdb at entry point
if options.qdb is True:
QlQdb(ql, rr=options.rr).run()
exit()
return ql
return ql_args
def handle_examples(parser: argparse.ArgumentParser):
@@ -211,7 +205,7 @@ if __name__ == '__main__':
code_parser.add_argument('-i', '--input', metavar="INPUT", dest="input", help='input hex value')
code_parser.add_argument('--arch', required=True, choices=arch_map, action=__arg_archtype)
code_parser.add_argument('--thumb', action='store_true', default=False, help='specify thumb mode for ARM')
code_parser.add_argument('--endian', choices=('little', 'big'), default='little', help='specify endianess for bi-endian archs')
code_parser.add_argument('--endian', choices=endian_map, default=QL_ENDIAN.EL, action=__arg_endian, help='specify endianess for bi-endian archs')
code_parser.add_argument('--os', required=True, choices=os_map, action=__arg_ostype)
code_parser.add_argument('--rootfs', default='.', help='emulated root filesystem, that is where all libraries reside')
code_parser.add_argument('--format', choices=('asm', 'hex', 'bin'), default='bin', help='input file format')
@@ -222,7 +216,6 @@ if __name__ == '__main__':
# set "qltui" subcommand
qltui_parser = commands.add_parser('qltui', help='show qiling Terminal User Interface', add_help=False)
qltui_enabled = False
comm_parser = run_parser
if len(sys.argv) > 1 and sys.argv[1] == 'code':
@@ -249,21 +242,44 @@ if __name__ == '__main__':
comm_parser.add_argument('--libcache', action='store_true', help='enable dll caching for windows')
options = parser.parse_args()
if options.subcommand == 'examples':
handle_examples(parser)
qltui_enabled = False
ql_args = {}
if options.subcommand == 'qltui':
import qltui
options = qltui.get_data()
qltui_enabled = True
if options.subcommand == 'examples':
handle_examples(parser)
# ql file setup
if options.subcommand == 'run':
ql = handle_run(options)
elif options.subcommand == 'run':
ql_args = handle_run(options)
# ql code setup
if options.subcommand == 'code':
ql = handle_code(options)
elif options.subcommand == 'code':
ql_args = handle_code(options)
ql_args.update({
'env': options.env,
'verbose': options.verbose,
'profile': options.profile,
'console': options.console,
'filter': options.filter,
'log_file': options.log_file,
'log_plain': options.log_plain,
'multithread': options.multithread,
'libcache': options.libcache
})
ql = Qiling(**ql_args)
# attach Qdb at entry point
if options.qdb:
QlQdb(ql, rr=bool(options.rr)).run()
parser.exit(0)
# ql execute additional options
if options.gdb:
@@ -281,20 +297,25 @@ if __name__ == '__main__':
ql.debug_stop = True
if options.root:
ql.root = True
ql.os.root = True
if qltui_enabled:
hook_dictionary = qltui.hook(ql)
# ql run
with cov_utils.collect_coverage(ql, options.coverage_format, options.coverage_file):
if qltui_enabled:
hook_dictionary = qltui.hook(ql)
if options.coverage_file:
with cov_utils.collect_coverage(ql, options.coverage_format, options.coverage_file):
ql.run(timeout=options.timeout)
else:
ql.run(timeout=options.timeout)
if options.json:
report = report.generate_report(ql)
rep = report.generate_report(ql)
if qltui_enabled:
report["syscalls"] = qltui.transform_syscalls(ql.os.stats.syscalls)
qltui.show_report(ql, report, hook_dictionary)
rep["syscalls"] = qltui.transform_syscalls(ql.os.stats.syscalls)
qltui.show_report(ql, rep, hook_dictionary)
else:
pprint(report)
pprint(rep)
exit(ql.os.exit_code)
tests/test_cpu_models.py
+67
View File
@@ -0,0 +1,67 @@
#!/usr/bin/env python3
#
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
import importlib
import unittest
from enum import Enum
from typing import Callable, Type
from qiling.arch.models import *
class CpuModelsTest(unittest.TestCase):
"""Make sure all *_CPU_MODEL enums are in sync with Unicorn's constants.
"""
@staticmethod
def __get_converter(ql_name: str) -> Callable[[str], str]:
"""Create a ql-name to uc-name convertion function.
"""
def wrapped(k: str) -> str:
arch, _, name = k.partition('_')
return ql_name.format(arch=arch, name=name)
return wrapped
def __test_cpu_models(self, uc_const_module: str, ql_models_enum: Type[Enum], uc_name: str):
ql_to_uc_name = self.__get_converter(uc_name)
uc_consts = importlib.import_module(f'unicorn.{uc_const_module}').__dict__
for k, v in ql_models_enum.__members__.items():
uc_const_name = ql_to_uc_name(k)
# make sure a ql enumeration entry has a cooresponding uc constant
self.assertIn(uc_const_name, uc_consts, f'Could not find a matching constant for {k} ({uc_const_name})')
# make sure their values are equal
self.assertEqual(uc_consts[uc_const_name], v.value, f'Unexpected value for {uc_const_name}')
def test_x86_cpu_models(self):
self.__test_cpu_models('x86_const', X86_CPU_MODEL, 'UC_CPU_X86_{name}')
def test_arm_cpu_models(self):
self.__test_cpu_models('arm_const', ARM_CPU_MODEL, 'UC_CPU_ARM_{name}')
def test_arm64_cpu_models(self):
self.__test_cpu_models('arm64_const', ARM64_CPU_MODEL, 'UC_CPU_ARM64_{name}')
def test_mips_cpu_models(self):
self.__test_cpu_models('mips_const', MIPS_CPU_MODEL, 'UC_CPU_MIPS32_{name}')
def test_ppc_cpu_models(self):
self.__test_cpu_models('ppc_const', PPC_CPU_MODEL, 'UC_CPU_PPC32_{name}')
def test_riscv_cpu_models(self):
self.__test_cpu_models('riscv_const', RISCV_CPU_MODEL, 'UC_CPU_RISCV32_{name}')
def test_riscv64_cpu_models(self):
self.__test_cpu_models('riscv_const', RISCV64_CPU_MODEL, 'UC_CPU_RISCV64_{name}')
if __name__ == "__main__":
unittest.main()
tests/test_elf.py
+254 -624
View File
File diff suppressed because it is too large Load Diff
tests/test_elf_multithread.py
+40 -40
View File
@@ -74,7 +74,7 @@ class ELFTest(unittest.TestCase):
if fd == 1:
content = ql.mem.read(write_buf, count)
logged.append(content.decode())
logged.extend(content.decode().splitlines())
ql = Qiling([fr'{X86_LINUX_ROOTFS}/bin/x86_multithreading'], X86_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
@@ -93,7 +93,7 @@ class ELFTest(unittest.TestCase):
if fd == 1:
content = ql.mem.read(write_buf, count)
logged.append(content.decode())
logged.extend(content.decode().splitlines())
ql = Qiling([fr'{ARM64_LINUX_ROOTFS}/bin/arm64_multithreading'], ARM64_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
@@ -112,7 +112,7 @@ class ELFTest(unittest.TestCase):
if fd == 1:
content = ql.mem.read(write_buf, count)
logged.append(content.decode())
logged.extend(content.decode().splitlines())
ql = Qiling([fr'{X64_LINUX_ROOTFS}/bin/x8664_multithreading'], X64_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
@@ -131,7 +131,7 @@ class ELFTest(unittest.TestCase):
if fd == 1:
content = ql.mem.read(write_buf, count)
logged.append(content.decode())
logged.extend(content.decode().splitlines())
ql = Qiling([fr'{MIPSEB_LINUX_ROOTFS}/bin/mips32_multithreading'], MIPSEB_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
@@ -150,7 +150,7 @@ class ELFTest(unittest.TestCase):
if fd == 1:
content = ql.mem.read(write_buf, count)
logged.append(content.decode())
logged.extend(content.decode().splitlines())
ql = Qiling([fr'{MIPSEL_LINUX_ROOTFS}/bin/mips32el_multithreading'], MIPSEL_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
@@ -169,7 +169,7 @@ class ELFTest(unittest.TestCase):
if fd == 1:
content = ql.mem.read(write_buf, count)
logged.append(content.decode())
logged.extend(content.decode().splitlines())
ql = Qiling([fr'{ARM_LINUX_ROOTFS}/bin/arm_multithreading'], ARM_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
@@ -189,7 +189,7 @@ class ELFTest(unittest.TestCase):
if fd == 1:
content = ql.mem.read(write_buf, count)
logged.append(content.decode())
logged.extend(content.decode().splitlines())
ql = Qiling([fr'{ARMEB_LINUX_ROOTFS}/bin/armeb_multithreading'], ARMEB_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
@@ -208,7 +208,7 @@ class ELFTest(unittest.TestCase):
if fd == 2:
content = ql.mem.read(write_buf, count)
logged.append(content.decode())
logged.extend(content.decode().splitlines())
ql = Qiling([fr'{X86_LINUX_ROOTFS}/bin/x86_tcp_test', '20000'], X86_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
@@ -224,8 +224,8 @@ class ELFTest(unittest.TestCase):
# 'server recv() return 14.\n'
# 'server send() 14 return 14.\n'
m = re.search(r'(?P<num>\d+)\.\s+\Z', logged[-2])
self.assertIsNotNone(m, 'could not extract numeric value from log message')
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
num = m.group('num')
msg = logged[-1].strip()
@@ -239,7 +239,7 @@ class ELFTest(unittest.TestCase):
if fd == 2:
content = ql.mem.read(write_buf, count)
logged.append(content.decode())
logged.extend(content.decode().splitlines())
ql = Qiling([fr'{X64_LINUX_ROOTFS}/bin/x8664_tcp_test', '20001'], X64_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
@@ -255,8 +255,8 @@ class ELFTest(unittest.TestCase):
# 'server recv() return 14.\n'
# 'server send() 14 return 14.\n'
m = re.search(r'(?P<num>\d+)\.\s+\Z', logged[-2])
self.assertIsNotNone(m, 'could not extract numeric value from log message')
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
num = m.group('num')
msg = logged[-1].strip()
@@ -270,7 +270,7 @@ class ELFTest(unittest.TestCase):
if fd == 2:
content = ql.mem.read(write_buf, count)
logged.append(content.decode())
logged.extend(content.decode().splitlines())
ql = Qiling([fr'{ARM_LINUX_ROOTFS}/bin/arm_tcp_test', '20002'], ARM_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
@@ -286,8 +286,8 @@ class ELFTest(unittest.TestCase):
# 'server read() return 14.\n'
# 'server write() 14 return 14.\n'
m = re.search(r'(?P<num>\d+)\.\s+\Z', logged[-2])
self.assertIsNotNone(m, 'could not extract numeric value from log message')
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
num = m.group('num')
msg = logged[-1].strip()
@@ -301,7 +301,7 @@ class ELFTest(unittest.TestCase):
if fd == 2:
content = ql.mem.read(write_buf, count)
logged.append(content.decode())
logged.extend(content.decode().splitlines())
ql = Qiling([fr'{ARM64_LINUX_ROOTFS}/bin/arm64_tcp_test', '20003'], ARM64_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
@@ -317,8 +317,8 @@ class ELFTest(unittest.TestCase):
# 'server recv() return 14.\n'
# 'server send() 14 return 14.\n'
m = re.search(r'(?P<num>\d+)\.\s+\Z', logged[-2])
self.assertIsNotNone(m, 'could not extract numeric value from log message')
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
num = m.group('num')
msg = logged[-1].strip()
@@ -332,7 +332,7 @@ class ELFTest(unittest.TestCase):
if fd == 2:
content = ql.mem.read(write_buf, count)
logged.append(content.decode())
logged.extend(content.decode().splitlines())
ql = Qiling([fr'{ARMEB_LINUX_ROOTFS}/bin/armeb_tcp_test', '20004'], ARMEB_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
@@ -348,8 +348,8 @@ class ELFTest(unittest.TestCase):
# 'server recv() return 14.\n'
# 'server send() 14 return 14.\n'
m = re.search(r'(?P<num>\d+)\.\s+\Z', logged[-2])
self.assertIsNotNone(m, 'could not extract numeric value from log message')
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
num = m.group('num')
msg = logged[-1].strip()
@@ -363,7 +363,7 @@ class ELFTest(unittest.TestCase):
if fd == 2:
content = ql.mem.read(write_buf, count)
logged.append(content.decode())
logged.extend(content.decode().splitlines())
ql = Qiling([fr'{MIPSEB_LINUX_ROOTFS}/bin/mips32_tcp_test', '20005'], MIPSEB_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
@@ -379,8 +379,8 @@ class ELFTest(unittest.TestCase):
# 'server recv() return 14.\n'
# 'server send() 14 return 14.\n'
m = re.search(r'(?P<num>\d+)\.\s+\Z', logged[-2])
self.assertIsNotNone(m, 'could not extract numeric value from log message')
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
num = m.group('num')
msg = logged[-1].strip()
@@ -394,7 +394,7 @@ class ELFTest(unittest.TestCase):
if fd == 2:
content = ql.mem.read(write_buf, count)
logged.append(content.decode())
logged.extend(content.decode().splitlines())
ql = Qiling([fr'{MIPSEL_LINUX_ROOTFS}/bin/mips32el_tcp_test', '20006'], MIPSEL_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
@@ -410,8 +410,8 @@ class ELFTest(unittest.TestCase):
# 'server read() return 14.\n'
# 'server write() 14 return 14.\n'
m = re.search(r'(?P<num>\d+)\.\s+\Z', logged[-2])
self.assertIsNotNone(m, 'could not extract numeric value from log message')
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
num = m.group('num')
msg = logged[-1].strip()
@@ -425,7 +425,7 @@ class ELFTest(unittest.TestCase):
if fd == 2:
content = ql.mem.read(write_buf, count)
logged.append(content.decode())
logged.extend(content.decode().splitlines())
ql = Qiling([fr'{X86_LINUX_ROOTFS}/bin/x86_udp_test', '20010'], X86_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
@@ -441,8 +441,8 @@ class ELFTest(unittest.TestCase):
# 'server recvfrom() return 14.\n'
# 'server sendto() 14 return 14.\n'
m = re.search(r'(?P<num>\d+)\.\s+\Z', logged[-2])
self.assertIsNotNone(m, 'could not extract numeric value from log message')
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
num = m.group('num')
msg = logged[-1].strip()
@@ -456,7 +456,7 @@ class ELFTest(unittest.TestCase):
if fd == 2:
content = ql.mem.read(write_buf, count)
logged.append(content.decode())
logged.extend(content.decode().splitlines())
ql = Qiling([fr'{X64_LINUX_ROOTFS}/bin/x8664_udp_test', '20011'], X64_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
@@ -472,8 +472,8 @@ class ELFTest(unittest.TestCase):
# 'server recvfrom() return 14.\n'
# 'server sendto() 14 return 14.\n'
m = re.search(r'(?P<num>\d+)\.\s+\Z', logged[-2])
self.assertIsNotNone(m, 'could not extract numeric value from log message')
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
num = m.group('num')
msg = logged[-1].strip()
@@ -487,7 +487,7 @@ class ELFTest(unittest.TestCase):
if fd == 2:
content = ql.mem.read(write_buf, count)
logged.append(content.decode())
logged.extend(content.decode().splitlines())
ql = Qiling([fr'{ARM64_LINUX_ROOTFS}/bin/arm64_udp_test', '20013'], ARM64_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
@@ -503,8 +503,8 @@ class ELFTest(unittest.TestCase):
# 'server recvfrom() return 14.\n'
# 'server sendto() 14 return 14.\n'
m = re.search(r'(?P<num>\d+)\.\s+\Z', logged[-2])
self.assertIsNotNone(m, 'could not extract numeric value from log message')
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
num = m.group('num')
msg = logged[-1].strip()
@@ -518,7 +518,7 @@ class ELFTest(unittest.TestCase):
if fd == 2:
content = ql.mem.read(write_buf, count)
logged.append(content.decode())
logged.extend(content.decode().splitlines())
ql = Qiling([fr'{ARMEB_LINUX_ROOTFS}/bin/armeb_udp_test', '20014'], ARMEB_LINUX_ROOTFS, multithread=True, verbose=QL_VERBOSE.DEBUG)
@@ -534,8 +534,8 @@ class ELFTest(unittest.TestCase):
# 'server recvfrom() return 14.\n'
# 'server sendto() 14 return 14.\n'
m = re.search(r'(?P<num>\d+)\.\s+\Z', logged[-2])
self.assertIsNotNone(m, 'could not extract numeric value from log message')
m = re.search(r'(?P<num>\d+)\.\Z', logged[-2])
self.assertIsNotNone(m, f'could not extract numeric value from log message "{logged[-2]}"')
num = m.group('num')
msg = logged[-1].strip()
tests/test_history.py
+80 -54
View File
@@ -1,100 +1,126 @@
import re
import unittest
from typing import List, Optional, Tuple
from qiling import Qiling
from qiling.const import QL_VERBOSE
from qiling.extensions.coverage.formats.history import History
from typing import List
class HistoryTest(unittest.TestCase):
P_LIBC = r".*libc(-\d\.\d+)?.so.*" # covers both generic (i.e. "libc.so.6") and specific (e.g. "libc-2.27.so")
P_LD = r"ld-linux.*"
@staticmethod
def sanitize_mmap_path(mmap: List[tuple]) -> List[tuple]:
'''Removes the path from the mmap tuple so that it can be compared to other mmaps
currently because the loader is handling loading ld and the main binary, we get the annotation of the path in element 5 of the tuple (index 4)
this path is going to be dependent on the users filesystem, so it doesnt quite make sense to test for it
'''
def sanitize_mmap_path(mmap: List[Tuple]) -> List[Tuple[int, int, str, str]]:
"""Removes the path from the mmap tuple so that it can be compared to other mmaps.
currently because the loader is handling loading ld and the main binary, we get the annotation of the path in
element 5 of the tuple (index 4) this path is going to be dependent on the users filesystem, so it doesnt quite
make sense to test for it
"""
if isinstance(mmap, tuple):
mmap = [mmap]
return list(map(lambda x: (x[0], x[1], x[2], x[3], ''), mmap))
return [tup[:4] for tup in mmap]
def setUp(self):
rootfs = '../examples/rootfs/x8664_linux'
argv = [fr'{rootfs}/bin/x8664_hello']
ql = Qiling(argv, rootfs, verbose=QL_VERBOSE.OFF)
self.history = History(ql)
self.ql = ql
def get_label(self, basename: str) -> Optional[str]:
"""Return the matching label from mapinfo, ignoring boxes and possible different suffixes.
For example, on some systems libc label may appear as "libc.so.6" and "libc-2.27.so" on others.
This method also ignores the boxed prefixes, if exists (e.g. "[mmap]")
"""
p = re.compile(rf'\A(\[.+\]\s+)?{basename}[.-]')
return next((label for _, _, _, label, _ in self.ql.mem.map_info if re.match(p, label)), None)
def test_get_regex_matching_exec_maps(self):
ql = Qiling(["../examples/rootfs/x8664_linux/bin/x8664_hello"], "../examples/rootfs/x8664_linux", verbose=QL_VERBOSE.OFF)
history = History(ql)
ql.run()
self.ql.run()
self.assertEqual([(0x7fffb7dd6000, 0x7fffb7fbd000, 'r-x', '[mmap] libc.so.6', '')], history.get_regex_matching_exec_maps(".*libc.so.*"))
self.assertEqual(
self.assertListEqual(
[
(0x7fffb7dd6000, 0x7fffb7fbd000, 'r-x', '[mmap] libc.so.6', ''),
(0x7ffff7dd5000, 0x7ffff7dfc000, 'r-x', 'ld-linux-x86-64.so.2', '')
(0x7fffb7dd6000, 0x7fffb7fbd000, 'r-x', self.get_label('libc'), '')
],
self.sanitize_mmap_path(history.get_regex_matching_exec_maps([".*libc.so.*", "ld.*"]))
self.history.get_regex_matching_exec_maps(self.P_LIBC)
)
del ql
self.assertListEqual(
[
(0x7fffb7dd6000, 0x7fffb7fbd000, 'r-x', self.get_label('libc')),
(0x7ffff7dd5000, 0x7ffff7dfc000, 'r-x', self.get_label('ld'))
],
self.sanitize_mmap_path(self.history.get_regex_matching_exec_maps([self.P_LIBC, self.P_LD]))
)
def test_get_mem_map_from_addr(self):
ql = Qiling(["../examples/rootfs/x8664_linux/bin/x8664_hello"], "../examples/rootfs/x8664_linux", verbose=QL_VERBOSE.OFF)
history = History(ql)
ql.run()
self.ql.run()
self.assertEqual(
self.sanitize_mmap_path(history.get_mem_map_from_addr(0x7ffff7df4830))[0],
(
0x7ffff7dd5000,
0x7ffff7dfc000,
'r-x',
'ld-linux-x86-64.so.2',
''))
mmap = self.history.get_mem_map_from_addr(0x7ffff7df4830)
self.assertIsNotNone(mmap)
self.assertTupleEqual(
(0x7ffff7dd5000, 0x7ffff7dfc000, 'r-x', self.get_label('ld')),
self.sanitize_mmap_path(mmap)[0]
)
def test_get_ins_exclude_lib(self):
ql = Qiling(["../examples/rootfs/x8664_linux/bin/x8664_hello"], "../examples/rootfs/x8664_linux", verbose=QL_VERBOSE.OFF)
history = History(ql)
ql.run(end=0x55555555465a)
self.ql.run(end=0x55555555465a)
non_libc_blocks = history.get_ins_exclude_lib(".*libc.so.*")
non_libc_blocks = self.history.get_ins_exclude_lib(self.P_LIBC)
self.assertGreater(len(non_libc_blocks), 0)
# this test is going to take a while but oh well
# also assumes that the get_mem_map_from_addr function works
for block in non_libc_blocks:
map_for_ins = history.get_mem_map_from_addr(block)
self.assertNotRegex(map_for_ins[3], ".*libc.so.*")
map_for_ins = self.history.get_mem_map_from_addr(block)
assert len(non_libc_blocks) > 0
self.assertIsNotNone(map_for_ins)
self.assertNotRegex(map_for_ins[3], self.P_LIBC)
non_libc_blocks_and_ld = history.get_ins_exclude_lib([".*libc.so.*", "ld-linux.*"])
non_libc_blocks_and_ld = self.history.get_ins_exclude_lib([self.P_LIBC, self.P_LD])
self.assertGreater(len(non_libc_blocks_and_ld), 0)
for block in non_libc_blocks_and_ld:
map_for_ins = history.get_mem_map_from_addr(block)
self.assertNotRegex(map_for_ins[3], ".*libc.so.*|ld-linux.*")
assert len(non_libc_blocks_and_ld) > 0
map_for_ins = self.history.get_mem_map_from_addr(block)
self.assertIsNotNone(map_for_ins)
self.assertNotRegex(map_for_ins[3], '|'.join((self.P_LIBC, self.P_LD)))
def test_get_ins_only_lib(self):
ql = Qiling(["../examples/rootfs/x8664_linux/bin/x8664_hello"], "../examples/rootfs/x8664_linux", verbose=QL_VERBOSE.OFF)
history = History(ql)
ql.run(end=0x55555555465a)
self.ql.run(end=0x55555555465a)
non_libc_blocks = history.get_ins_only_lib(".*libc.so.*")
non_libc_blocks = self.history.get_ins_only_lib(self.P_LIBC)
self.assertGreater(len(non_libc_blocks), 0)
# this test is going to take a while but oh well
# also assumes that the get_mem_map_from_addr function works
for block in non_libc_blocks:
map_for_ins = history.get_mem_map_from_addr(block)
self.assertRegex(map_for_ins[3], ".*libc.so.*")
map_for_ins = self.history.get_mem_map_from_addr(block)
assert len(non_libc_blocks) > 0
self.assertIsNotNone(map_for_ins)
self.assertRegex(map_for_ins[3], self.P_LIBC)
non_libc_blocks_and_ld = history.get_ins_only_lib([".*libc.so.*", "ld-linux.*"])
non_libc_blocks_and_ld = self.history.get_ins_only_lib([self.P_LIBC, self.P_LD])
self.assertGreater(len(non_libc_blocks_and_ld), 0)
for block in non_libc_blocks_and_ld :
map_for_ins = history.get_mem_map_from_addr(block)
self.assertRegex(map_for_ins[3], ".*libc.so.*|.*ld-linux.*")
for block in non_libc_blocks_and_ld:
map_for_ins = self.history.get_mem_map_from_addr(block)
self.assertIsNotNone(map_for_ins)
self.assertRegex(map_for_ins[3], '|'.join((self.P_LIBC, self.P_LD)))
assert len(non_libc_blocks_and_ld) > 0
if __name__ == "__main__":
unittest.main()
unittest.main()
tests/test_qltool.py
+26 -29
View File
@@ -5,50 +5,47 @@
import os
import subprocess
import re
import sys
import unittest
class Qltool_Test(unittest.TestCase):
def test_qltool_exec_args(self):
create = [sys.executable, '../qltool', 'run', '-f', '../examples/rootfs/x8664_linux/bin/x8664_args', '--rootfs', '../examples/rootfs/x8664_linux', '--verbose', 'off', '--args', 'test1', 'test2' ,'test3']
p = subprocess.Popen(create, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
for line in iter(p.stdout.readline, b''):
self.stdout = line
def __run(self, cmdline: str) -> bytes:
try:
output = subprocess.check_output([sys.executable] + cmdline.split(), stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
raise RuntimeError(f"command '{e.cmd}' return with error (code {e.returncode}): {e.output}")
else:
return output
self.assertEqual(b'arg 2 test3\n', self.stdout)
def test_qltool_exec_args(self):
output = self.__run(r'../qltool run -f ../examples/rootfs/x8664_linux/bin/x8664_args --rootfs ../examples/rootfs/x8664_linux --verbose off --args test1 test2 test3')
self.assertEqual(b'arg 2 test3', output.splitlines()[-1])
def test_qltool_shellcode(self):
create = [sys.executable, '../qltool', 'code', '--os','linux','--arch', 'x86', '--format', 'asm', '-f', '../examples/shellcodes/lin32_execve.asm']
try:
subprocess.check_output(create,stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
raise RuntimeError("command '{}' return with error (code {}): {}".format(e.cmd, e.returncode, e.output))
self.__run(r'../qltool code --os linux --arch x86 --format asm -f ../examples/shellcodes/lin32_execve.asm')
def test_qltool_coverage(self):
os.makedirs("./log_test", exist_ok=True)
create = [sys.executable, '../qltool', 'run', '-f','../examples/rootfs/x8664_efi/bin/TcgPlatformSetupPolicy','--rootfs', '../examples/rootfs/x8664_efi','--coverage-format', 'drcov', '--coverage-file', 'log_test/TcgPlatformSetupPolicy']
try:
subprocess.check_output(create, stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
raise RuntimeError("command '{}' return with error (code {}): {}".format(e.cmd, e.returncode, e.output))
os.makedirs(r'./log_test', exist_ok=True)
self.__run(r'../qltool run -f ../examples/rootfs/x8664_efi/bin/TcgPlatformSetupPolicy --rootfs ../examples/rootfs/x8664_efi --coverage-format drcov --coverage-file log_test/TcgPlatformSetupPolicy')
def test_qltool_json(self):
create = [sys.executable, '../qltool', 'run', '-f','../examples/rootfs/x86_linux/bin/x86_hello','--rootfs', '../examples/rootfs/x86_linux','--verbose', 'off', '--json']
try:
subprocess.check_output(create, stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
raise RuntimeError("command '{}' return with error (code {}): {}".format(e.cmd, e.returncode, e.output))
self.__run(r'../qltool run -f ../examples/rootfs/x86_linux/bin/x86_hello --rootfs ../examples/rootfs/x86_linux --verbose off --json')
def test_qltool_filter(self):
create = [sys.executable, '../qltool', 'run', '-f', '../examples/rootfs/arm_linux/bin/arm_hello', '--rootfs', '../examples/rootfs/arm_linux', '-e', '^(open|brk)', '--log-plain']
try:
output = subprocess.check_output(create, stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
raise RuntimeError("command '{}' return with error (code {}): {}".format(e.cmd, e.returncode, e.output))
filter_pattern = r'^(open|brk)'
output = self.__run(fr'../qltool run -f ../examples/rootfs/arm_linux/bin/arm_hello --rootfs ../examples/rootfs/arm_linux -e {filter_pattern} --log-plain')
lines = [ line.strip('[=]\t') for line in output.decode().split("\n")]
self.assertTrue(all(filter(lambda x: x.startswith("open") or x.startswith("brk"), lines)))
# keep only log entries and strip them from log prefix
p = re.compile(rb'^\[\S\]\s+')
log_entries = (p.sub(b'', line) for line in output.splitlines() if p.match(line))
# make sure that all log entries are of the expected regex filter pattern
p = re.compile(filter_pattern.encode())
self.assertTrue(all(p.search(entry) is not None for entry in log_entries))
if __name__ == "__main__":
tests/test_struct.py
+143 -88
View File
@@ -3,17 +3,19 @@
# Cross Platform and Multi Architecture Advanced Binary Emulation Framework
#
import sys, unittest
import ctypes
import unittest
from typing import Optional
import sys
sys.path.append("..")
import ctypes
from qiling import Qiling
from qiling.const import QL_ARCH, QL_OS
from qiling.os.struct import BaseStruct
class DummyInternalStruct(BaseStruct):
_fields_ = [
('X', ctypes.c_uint32)
@@ -35,125 +37,178 @@ class DummyStruct(BaseStruct):
# we only need context and not going to run anything anyway, so just use whatever
NOPSLED = b'\x90' * 8
ROOTFS = r'../examples/rootfs/x8664_linux'
class StructTest(unittest.TestCase):
def setUp(self) -> None:
ql = Qiling(code=NOPSLED, rootfs=ROOTFS, archtype=QL_ARCH.X8664, ostype=QL_OS.LINUX)
self.ptr = 0x100000
self.mem = ql.mem
self.expected = {
'A' : 0xdeadface,
'B' : 0x1020304050607080,
'C' : DummyInternalStruct(0x11213141),
'D' : b'Hello World!',
}
# create a dummy structure with expected values
dummy = DummyStruct(**self.expected)
# emit dummy structure to memory
ql.mem.map(self.ptr, ql.mem.align_up(dummy.sizeof()))
ql.mem.write(self.ptr, bytes(dummy))
def __read_data(self, offset: int = 0, size: Optional[int] = None) -> bytearray:
return self.mem.read(self.ptr + offset, size or DummyStruct.sizeof())
class Wrapper:
class StructTest(unittest.TestCase):
OSTYPE: QL_OS
ROOTFS: str
def setUp(self) -> None:
ql = Qiling(code=NOPSLED, rootfs=self.ROOTFS, archtype=QL_ARCH.X8664, ostype=self.OSTYPE)
def __write_data(self, offset: int, data: bytes) -> None:
self.mem.write(self.ptr + offset, data)
self.ptr = 0x100000
self.mem = ql.mem
self.expected = {
'A': 0xdeadface,
'B': 0x1020304050607080,
'C': DummyInternalStruct(0x11213141),
'D': b'Hello World!',
}
@staticmethod
def __to_uint(data: bytearray) -> int:
return int.from_bytes(data, 'little', signed=False)
# create a dummy structure with expected values
dummy = DummyStruct(**self.expected)
# emit dummy structure to memory
ql.mem.map(self.ptr, ql.mem.align_up(dummy.sizeof()))
ql.mem.write(self.ptr, bytes(dummy))
def test_load_from(self):
dummy = DummyStruct.load_from(self.mem, self.ptr)
def __read_data(self, offset: int = 0, size: Optional[int] = None) -> bytearray:
return self.mem.read(self.ptr + offset, size or DummyStruct.sizeof())
self.assertEqual(self.expected['A'], dummy.A)
self.assertEqual(self.expected['B'], dummy.B)
self.assertEqual(self.expected['C'], dummy.C)
self.assertEqual(self.expected['D'], dummy.D)
def __write_data(self, offset: int, data: bytes) -> None:
self.mem.write(self.ptr + offset, data)
@staticmethod
def __to_uint(data: bytearray) -> int:
return int.from_bytes(data, 'little', signed=False)
def test_save_to(self):
dummy = DummyStruct(
A=0x0c0a0f0e,
B=0x1828384858687888,
C=DummyInternalStruct(0x19293949),
D=b'Goodbye World!'
)
def test_load_from(self):
dummy = DummyStruct.load_from(self.mem, self.ptr)
dummy.save_to(self.mem, self.ptr)
self.assertEqual(self.expected['A'], dummy.A)
self.assertEqual(self.expected['B'], dummy.B)
self.assertEqual(self.expected['C'], dummy.C)
self.assertEqual(self.expected['D'], dummy.D)
obj_data = bytes(dummy)
mem_data = self.__read_data()
def test_save_to(self):
dummy = DummyStruct(
A=0x0c0a0f0e,
B=0x1828384858687888,
C=DummyInternalStruct(0x19293949),
D=b'Goodbye World!'
)
self.assertEqual(obj_data, mem_data)
dummy.save_to(self.mem, self.ptr)
def test_ref_discard(self):
data_before = self.__read_data()
obj_data = bytes(dummy)
mem_data = self.__read_data()
unused = []
with DummyStruct.ref(self.mem, self.ptr) as dummy:
print(f'B = {dummy.B:#x}')
print(f'C = {dummy.C}')
self.assertEqual(obj_data, mem_data)
unused.append(dummy.A + 1337)
def test_ref_discard(self):
data_before = self.__read_data()
data_after = self.__read_data()
unused = []
with DummyStruct.ref(self.mem, self.ptr) as dummy:
print(f'B = {dummy.B:#x}')
print(f'C = {dummy.C}')
self.assertEqual(data_before, data_after)
unused.append(dummy.A + 1337)
def test_ref_save(self):
expected = 0x10303070
data_after = self.__read_data()
with DummyStruct.ref(self.mem, self.ptr) as dummy:
print(f'B = {dummy.B:#x}')
print(f'C = {dummy.C}')
self.assertEqual(data_before, data_after)
def test_ref_save(self):
expected = 0x10303070
with DummyStruct.ref(self.mem, self.ptr) as dummy:
print(f'B = {dummy.B:#x}')
print(f'C = {dummy.C}')
dummy.A = expected
data = self.__read_data(DummyStruct.offsetof('A'), 4)
self.assertEqual(expected, Wrapper.StructTest.__to_uint(data))
def test_ref_save_internal(self):
expected = 0x16363676
with DummyStruct.ref(self.mem, self.ptr) as dummy:
dummy.C.X = expected
data = self.__read_data(DummyStruct.offsetof('C') + DummyInternalStruct.offsetof('X'), 4)
self.assertEqual(expected, Wrapper.StructTest.__to_uint(data))
def test_volatile_ref(self):
dummy = DummyStruct.volatile_ref(self.mem, self.ptr)
expected = 0x01030307
dummy.A = expected
data = self.__read_data(DummyStruct.offsetof('A'), 4)
self.assertEqual(expected, Wrapper.StructTest.__to_uint(data))
data = self.__read_data(DummyStruct.offsetof('A'), 4)
self.assertEqual(expected, StructTest.__to_uint(data))
self.assertEqual(self.expected['B'], dummy.B)
self.assertEqual(self.expected['C'], dummy.C)
def test_ref_save_internal(self):
expected = 0x16363676
expected = b'Volatility Test!'
self.__write_data(DummyStruct.offsetof('D'), expected)
self.assertEqual(expected, dummy.D)
with DummyStruct.ref(self.mem, self.ptr) as dummy:
def test_volatile_ref_internal(self):
dummy = DummyStruct.volatile_ref(self.mem, self.ptr)
expected = 0x51535357
dummy.C.X = expected
data = self.__read_data(DummyStruct.offsetof('C') + DummyInternalStruct.offsetof('X'), 4)
self.assertEqual(expected, Wrapper.StructTest.__to_uint(data))
data = self.__read_data(DummyStruct.offsetof('C') + DummyInternalStruct.offsetof('X'), 4)
self.assertEqual(expected, StructTest.__to_uint(data))
# BUG: ctype.c_wchar size differs between python for windows and linux.on windows it has the
# size of 2 as expected (same as the 'utf-16le' encoding), where on linux the size is 4 (same
# as the 'utf-32' encoding), which expands the size of the field and doesn't provide the
# expected outcome.
#
# currently wchar is used only in windows structres, which makes it problematic to emulate
# windows programs on top of a linux host. until we implement our own wchar which depends on
# the emualted os rather than the hosting os, the following tests are conidered broken.
def test_volatile_ref(self):
dummy = DummyStruct.volatile_ref(self.mem, self.ptr)
class WCharTest(unittest.TestCase):
WCHAR_ENC: str
expected = 0x01030307
dummy.A = expected
data = self.__read_data(DummyStruct.offsetof('A'), 4)
self.assertEqual(expected, StructTest.__to_uint(data))
@unittest.skip('broken wchar implementation')
def test_wchar(self):
wchar_enc = self.WCHAR_ENC
wchar_size = len('\x00'.encode(wchar_enc))
self.assertEqual(self.expected['B'], dummy.B)
self.assertEqual(self.expected['C'], dummy.C)
field_type = ctypes.c_wchar # TODO: replace ctypes.c_wchar with an alternate implementation
field_data = 'A'
self.assertEqual(len(field_data) * wchar_size, ctypes.sizeof(field_type))
expected = b'Volatility Test!'
self.__write_data(DummyStruct.offsetof('D'), expected)
self.assertEqual(expected, dummy.D)
obj = field_type(field_data)
self.assertEqual(field_data, obj.value)
def test_volatile_ref_internal(self):
dummy = DummyStruct.volatile_ref(self.mem, self.ptr)
obj_bytes = bytes(obj)
self.assertEqual(field_data.encode(wchar_enc), obj_bytes)
expected = 0x51535357
dummy.C.X = expected
data = self.__read_data(DummyStruct.offsetof('C') + DummyInternalStruct.offsetof('X'), 4)
self.assertEqual(expected, StructTest.__to_uint(data))
@unittest.skip('broken wchar implementation')
def test_wchar_array(self):
wchar_enc = self.WCHAR_ENC
wchar_size = len('\x00'.encode(wchar_enc))
field_elems = 16
field_type = ctypes.c_wchar * field_elems # TODO: replace ctypes.c_wchar with an alternate implementation
field_data = 'unicode sucks'
self.assertEqual(field_elems * wchar_size, ctypes.sizeof(field_type))
obj = field_type(*field_data)
self.assertEqual(field_data, obj.value)
obj_bytes = bytes(obj)
self.assertEqual(field_data.ljust(field_elems, '\x00').encode(wchar_enc), obj_bytes)
class StructTestLinux(Wrapper.StructTest, Wrapper.WCharTest):
OSTYPE = QL_OS.LINUX
ROOTFS = r'../examples/rootfs/x8664_linux'
WCHAR_ENC = 'utf-32le'
class StructTestWindows(Wrapper.StructTest, Wrapper.WCharTest):
OSTYPE = QL_OS.WINDOWS
ROOTFS = r'../examples/rootfs/x8664_windows'
WCHAR_ENC = 'utf-16le'
if __name__ == "__main__":