520 lines
No EOL
19 KiB
Python
520 lines
No EOL
19 KiB
Python
#!/usr/bin/env python
|
|
# Sample code for ARM of Unicorn. Nguyen Anh Quynh <aquynh@gmail.com>
|
|
# Python sample ported by Loi Anh Tuan <loianhtuan@gmail.com>
|
|
# pyright: reportUnknownMemberType=false, reportUnusedCallResult=false, reportAny=false
|
|
|
|
from __future__ import print_function
|
|
from unicorn import *
|
|
from unicorn.arm_const import *
|
|
from capstone import *
|
|
from capstone.arm import *
|
|
import crcmod
|
|
from typing import Any
|
|
import time
|
|
|
|
mmios = {}
|
|
|
|
class MMIOMetaClass(type):
|
|
|
|
def __new__(mcs, name, bases, attrs):
|
|
cls = type.__new__(mcs, name, bases, attrs)
|
|
#print("Create: %r: %s" % (cls, name))
|
|
if name != 'MMIO':
|
|
mmios[name.lower()] = cls
|
|
return cls
|
|
|
|
class MMIO(object):
|
|
"""
|
|
MMIO devices are subclassed from MMIO.
|
|
A MMIO device has an address and size, and will be called through read() and write()
|
|
functions when data aborts due to read and writes being made to the addresses within
|
|
that range.
|
|
The metaclass is used to track all the MMIO objects created in the global 'mmios'
|
|
dictionary.
|
|
"""
|
|
__metaclass__ = MMIOMetaClass
|
|
address = None
|
|
size = None
|
|
|
|
def __init__(self, ro):
|
|
self.ro = ro
|
|
|
|
def __repr__(self):
|
|
return "<MMIO:%s>" % (self.__class__.__name__,)
|
|
|
|
def initialise(self):
|
|
pass
|
|
|
|
def read(self, offset, size):
|
|
return 0
|
|
|
|
def write(self, offset, size, value):
|
|
pass
|
|
|
|
class MMIODisabledError(Exception):
|
|
"""
|
|
Raised by MMIOProcess to indicate that we're not doing memory mapped I/O.
|
|
"""
|
|
pass
|
|
|
|
class MMIOs(object):
|
|
"""
|
|
MMIO region dispatcher for an emulator instance.
|
|
An emulator instance ('ro' in the class) creates the MMIOs object, and performs all operations
|
|
on those MMIO objects through it. When the emulator is to be started for the first time, this
|
|
is the equivalent of hardware being powered, so the 'initialise' method should be called. If
|
|
a reset of the system is required, this initialise can be called again. This will construct
|
|
all the MMIO objects declared by the system.
|
|
When a data abort occurs, this is dispatched to the unmapped_access method, which will either
|
|
handle it and return True, or return False if the region is not known.
|
|
"""
|
|
|
|
def __init__(self, ro):
|
|
self.ro = ro
|
|
self.mmioprocess = MMIOProcess(ro=self.ro)
|
|
self.enabled = True
|
|
self.mmios = {}
|
|
self.regions = []
|
|
|
|
self.debug_mmio = False
|
|
self.ro.debug_register_ivar('mmio', self)
|
|
|
|
def __repr__(self):
|
|
return "<MMIOs(enable=%r, %s regions registered)>" % (self.enabled, len(self.regions))
|
|
|
|
def __getitem__(self, key):
|
|
return self.mmios[key.lower()]
|
|
|
|
def initialise(self):
|
|
# Construct an object per registered resource
|
|
self.mmios = dict((name, cls(self.ro)) for name, cls in mmios.items())
|
|
self.regions = sorted([(mmio.address, mmio.address + mmio.size, mmio) for mmio in self.mmios.values()])
|
|
if self.debug_mmio:
|
|
print("Regions = %r" % (self.regions,))
|
|
for mmio in self.mmios.values():
|
|
mmio.initialise()
|
|
|
|
def unmapped_access(self, address, size, read=True):
|
|
mmio = None
|
|
for block in self.regions:
|
|
if block[1] < address:
|
|
# Not handled (this is an ordered list, so this entry cannot be present
|
|
return False
|
|
|
|
if address >= block[0] and address < block[1]:
|
|
mmio = block[2]
|
|
break
|
|
|
|
if not mmio:
|
|
return False
|
|
|
|
if self.debug_mmio:
|
|
print("Unmapped access to %08x, read=%r (mmio=%r)" % (address, read, mmio))
|
|
|
|
if read:
|
|
read_func = lambda address, size: mmio.read(address - mmio.address, size)
|
|
write_func = None
|
|
else:
|
|
read_func = None
|
|
write_func = lambda address, size, data: mmio.write(address - mmio.address, size, data)
|
|
try:
|
|
self.mmioprocess.trap(address, size, read_func, write_func)
|
|
except Exception as exc:
|
|
# Any failure in MMIOProcess means that we didn't handle this instruction, and will
|
|
# abort.
|
|
return False
|
|
return True
|
|
|
|
class MMIOProcess(object):
|
|
pagesize = 4096
|
|
|
|
def __init__(self, ro):
|
|
self.ro = ro
|
|
self.address = None
|
|
self.read = None
|
|
self.write = None
|
|
self.size = None
|
|
self.page = None
|
|
self.hook = None
|
|
self.hook_next = False
|
|
self.enabled = True
|
|
|
|
self.debug_mmioprocess = False
|
|
self.ro.debug_register_ivar('mmioprocess', self)
|
|
|
|
@property
|
|
def enabled(self):
|
|
return self.hook is not None
|
|
|
|
@enabled.setter
|
|
def enabled(self, value):
|
|
if value:
|
|
if self.hook is None:
|
|
self.hook = self.ro.emu.hook_add(unicorn.UC_HOOK_CODE, self.code_hook, begin=0, end=0xFFFFFFFF)
|
|
else:
|
|
if self.hook:
|
|
self.ro.emu.hook_del(self.hook)
|
|
self.hook = None
|
|
|
|
def trap(self, address, size, read=None, write=None):
|
|
if not self.enabled:
|
|
raise MMIODisabledError("MMIO is disabled")
|
|
|
|
self.address = address
|
|
self.read = read
|
|
self.write = write
|
|
self.size = size
|
|
self.page = address & ~(self.pagesize - 1)
|
|
hook_pc = self.ro.regs.pc
|
|
if self.debug_mmioprocess:
|
|
print("Hook pc = %08x" % (hook_pc,))
|
|
print("Mapping page %08x" % (self.page,))
|
|
self.ro.emu.mem_map(self.page, self.pagesize)
|
|
if read:
|
|
data = self.read(self.address, size)
|
|
self.ro.memory.write_word(data, offset=address)
|
|
self.hook_next = True
|
|
|
|
def code_hook(self, uc, address, size, user_data):
|
|
if not self.hook_next:
|
|
return
|
|
self.hook_next = False
|
|
|
|
if self.debug_mmioprocess:
|
|
print("Post trapped with instruction at %08x" % (address,))
|
|
if self.write:
|
|
data = self.ro.memory.read_word(self.address)
|
|
if self.debug_mmioprocess:
|
|
print("Read data %08x" % (data,))
|
|
self.write(self.address, self.size, data)
|
|
|
|
if self.debug_mmioprocess:
|
|
print("Unmap %08x" % (self.page,))
|
|
uc.mem_unmap(self.page, self.pagesize)
|
|
|
|
class CounterMMIO(MMIO):
|
|
"""
|
|
This counter class is an example to show how a MMIO device can be implemented.
|
|
"""
|
|
address = 0x4000
|
|
size = 0x100
|
|
|
|
def __init__(self, ro):
|
|
super(CounterMMIO, self).__init__(ro)
|
|
self.counters = [0] * self.size
|
|
|
|
def read(self, offset, size):
|
|
value = self.counters[offset]
|
|
self.counters[offset] += 1
|
|
return value
|
|
|
|
def write(self, offset, size, value):
|
|
self.counters[offset] = value
|
|
|
|
DEBUG_INFO = True
|
|
DEBUG = True
|
|
|
|
# callback for tracing basic blocks
|
|
# def hook_block(uc, address, size, user_data):
|
|
# pass
|
|
# #print(">>> Tracing basic block at 0x%x, block size = 0x%x" %(address, size))
|
|
|
|
# status_reg = 0b0100 << 28 # By default, the DCC loader can write, but not read
|
|
# rd_reg = 0
|
|
# wr_reg = 0
|
|
|
|
# callback for tracing instructions
|
|
|
|
# cp = 15
|
|
# is64 = 0
|
|
# sec = 0
|
|
# crn = 1
|
|
# crm = 0
|
|
# opc1 = 0
|
|
# opc2 = 0
|
|
# val = ??
|
|
|
|
# Test ARM
|
|
class DCCTarget():
|
|
def __init__(self, load_offset: int, loader: bytes | bytearray, ram_size: int):
|
|
self.__status_reg: int = 0b0100 << 28 # By default, the DCC loader can write, but not read
|
|
self.__rd_reg: int = 0
|
|
self.__wr_reg: int = 0
|
|
self.__load_offset = load_offset
|
|
self.__loader_data: bytes = bytes(loader)
|
|
|
|
self.emu: Uc = Uc(UC_ARCH_ARM, UC_MODE_ARM)
|
|
self.emu.mem_map(load_offset, ram_size)
|
|
self.emu.mem_write(load_offset, self.__loader_data)
|
|
self.emu.reg_write(UC_ARM_REG_APSR, 0xFFFFFFFF)
|
|
|
|
# tracing one instruction at ADDRESS with customized callback
|
|
self.emu.hook_add(UC_HOOK_CODE, self._dcc_handle_code)
|
|
|
|
def on_read(mu: Uc, access: int, address: int, size: int, value: int, user_data: Any):
|
|
if DEBUG and DEBUG_INFO:
|
|
print("Read at", hex(address), size, self.emu.mem_read(address, size))
|
|
|
|
def on_write(mu, access: int, address: int, size: int, value: int, user_data: Any):
|
|
if DEBUG and DEBUG_INFO:
|
|
print("Write at", hex(address), size, hex(value))
|
|
|
|
def on_error(mu, access: int, address: int, size: int, value: int, user_data: Any):
|
|
if DEBUG:
|
|
print("Error at", hex(address), size, hex(value), "in", hex(self.emu.reg_read(UC_ARM_REG_PC)), "lr", hex(self.emu.reg_read(UC_ARM_REG_LR)))
|
|
|
|
self.emu.hook_add(UC_HOOK_MEM_READ, on_read)
|
|
self.emu.hook_add(UC_HOOK_MEM_WRITE, on_write)
|
|
self.emu.hook_add(UC_HOOK_MEM_INVALID, on_error)
|
|
|
|
def run(self):
|
|
self.emu.emu_start(self.__load_offset, 0xffffffff)
|
|
|
|
def _dcc_handle_code(self, uc: Uc, address: int, size: int, user_data: Any):
|
|
try:
|
|
instr = self.__loader_data[address - self.__load_offset:(address - self.__load_offset) + size]
|
|
|
|
if DEBUG and DEBUG_INFO and False:
|
|
print(">>> Tracing instruction at 0x%x, instruction size = 0x%x" %(address, size))
|
|
print("CODE:", instr)
|
|
print("RSP1", hex(uc.reg_read(UC_ARM_REG_R0)))
|
|
print("RSP2", hex(uc.reg_read(UC_ARM_REG_R1)))
|
|
print("RSP3", hex(uc.reg_read(UC_ARM_REG_R2)))
|
|
print("RSP4", hex(uc.reg_read(UC_ARM_REG_R3)))
|
|
print("RLOC", hex(uc.reg_read(UC_ARM_REG_R9)))
|
|
print("SP", hex(uc.reg_read(UC_ARM_REG_SP)))
|
|
#print("SP_DATA", uc.mem_read(uc.reg_read(UC_ARM_REG_SP), 0x10))
|
|
|
|
instr_int = int.from_bytes(instr, "little")
|
|
|
|
if ((instr_int >> 24) & 0xf) != 0b1110:
|
|
return
|
|
|
|
print(hex(instr_int), hex(address))
|
|
|
|
cs = Cs(CS_ARCH_ARM, CS_MODE_THUMB if uc.reg_read(UC_ARM_REG_CPSR) & (1 << 5) else CS_MODE_ARM)
|
|
cs.detail = True
|
|
|
|
ins: CsInsn = [x for x in cs.disasm(instr, address)][0]
|
|
if ins.id == ARM_INS_MRC: # Read Coprocessor
|
|
if ins.operands[0].value.imm == 14:
|
|
opc_1 = ins.operands[1].value.imm
|
|
cp_dest = ins.operands[2].value.reg
|
|
cr_n = ins.operands[3].value.imm
|
|
cr_m = ins.operands[4].value.imm
|
|
opc_2 = ins.operands[5].value.imm
|
|
|
|
if cr_n == 0:
|
|
uc.reg_write(cp_dest, self.__status_reg)
|
|
|
|
elif cr_n == 1:
|
|
print("DCC HOST -> OCD", hex(self.__rd_reg))
|
|
uc.reg_write(cp_dest, self.__rd_reg)
|
|
self.__status_reg &= ~1 # Sets the R bit to low, indicating that the host has finished processing the data.
|
|
|
|
uc.reg_write(UC_ARM_REG_PC, address+size) # Skip this instruction as we've already processed some DCC logic
|
|
|
|
elif ins.id == ARM_INS_MCR: # Write Coprocessor
|
|
if ins.operands[0].value.imm == 14:
|
|
opc_1 = ins.operands[1].value.imm
|
|
cp_dest = ins.operands[2].value.reg
|
|
cr_n = ins.operands[3].value.imm
|
|
cr_m = ins.operands[4].value.imm
|
|
opc_2 = ins.operands[5].value.imm
|
|
|
|
if cr_n == 0:
|
|
self.__status_reg = uc.reg_read(cp_dest)
|
|
|
|
elif cr_n == 1:
|
|
self.__wr_reg = uc.reg_read(cp_dest)
|
|
print("DCC OCD -> HOST", hex(self.__wr_reg))
|
|
self.__status_reg |= 2 # Sets the W bit to high, indicating that the debugger is ready to process the data.
|
|
|
|
uc.reg_write(UC_ARM_REG_PC, address+size) # Skip this instruction as we've already processed some DCC logic
|
|
|
|
except KeyboardInterrupt:
|
|
uc.emu_stop()
|
|
raise
|
|
|
|
def dcc_read(self) -> int:
|
|
while (self.__status_reg & 2) == 0:
|
|
time.sleep(0.1)
|
|
|
|
temp = self.__wr_reg
|
|
|
|
self.__status_reg &= ~2 # Debugger finally processed the data and set the W bit to low.
|
|
return temp
|
|
|
|
def dcc_write(self, data: int):
|
|
while self.__status_reg & 1:
|
|
time.sleep(0.1)
|
|
|
|
self.__rd_reg = data
|
|
self.__status_reg |= 1 # With the R bit set to high, the host was as motivated to process the data.
|
|
|
|
# def test_arm():
|
|
# print("Emulate ARM code")
|
|
# try:
|
|
# # Initialize emulator in ARM mode
|
|
# mu = Uc(UC_ARCH_ARM, UC_MODE_ARM)
|
|
# mu.ctl_exits_enabled(True)
|
|
# mu.ctl_set_exits([0])
|
|
|
|
# mu.mem_map(0x00000000, 32 * 1024 * 1024)
|
|
# mu.mem_map(0x12000000, 32 * 1024 * 1024)
|
|
# mu.mem_map(0x03000000, 2 * 1024 * 1024)
|
|
|
|
# # map 2MB memory for this emulation
|
|
# mu.mem_map(0x14000000, 2 * 1024 * 1024)
|
|
|
|
# # write machine code to be emulated to memory
|
|
# mu.mem_write(0x14000000, open(DCC_LOADER, "rb").read())
|
|
# #mu.mem_write(0x00000000, open("cfi_32mb.bin", "rb").read())
|
|
# #mu.mem_write(0x00000000, b"\x01\x00\x7e\x22") # Infineon NOR
|
|
# #mu.mem_write(0x14000020, b"\x00\x00\x00\x00") # Infineon NOR
|
|
# #mu.mem_write(0x00000020, b"Q\0R\0Y\0\x02\0\0\0")
|
|
# #mu.mem_write(0x0000004e, (23).to_bytes(2, "little"))
|
|
|
|
# #mu.mem_write(0x14000020, b"\0\0\0\x10")
|
|
|
|
# # initialize machine registers
|
|
# mu.reg_write(UC_ARM_REG_APSR, 0xFFFFFFFF) #All application flags turned on
|
|
|
|
# # tracing all basic blocks with customized callback
|
|
# mu.hook_add(UC_HOOK_BLOCK, hook_block)
|
|
|
|
# # tracing one instruction at ADDRESS with customized callback
|
|
# mu.hook_add(UC_HOOK_CODE, hook_code)
|
|
|
|
# def on_read(mu, access, address, size, value, data):
|
|
# #if DEBUG and address <= 0x14000000:
|
|
# if DEBUG and DEBUG_INFO:
|
|
# print("Read at", hex(address), size, mu.mem_read(address, size))
|
|
|
|
# def on_write(mu, access, address, size, value, data):
|
|
# if DEBUG:
|
|
# if address <= 0x14000000:
|
|
# if (address & 0x1ffff) == 0xaaa and value == 0x98:
|
|
# mu.mem_write(0x00000000, open("cfi_32mb.bin", "rb").read())
|
|
# mu.mem_write(0x12000000, open("cfi_32mb.bin", "rb").read())
|
|
|
|
# elif (address & 0x1ffff) == 0xaaa and value == 0x90:
|
|
# mu.mem_write(0x00000000, b"\x01\x00\x7e\x22")
|
|
# mu.mem_write(0x12000000, b"\x01\x00\x7e\x22")
|
|
|
|
# elif (address & 0x1ffff) == 0x0 and value == 0xf0:
|
|
# mu.mem_write(0x00000000, open(DCC_FW, "rb").read())
|
|
# mu.mem_write(0x12000000, open(DCC_FW, "rb").read())
|
|
# # mu.reg_write(0x)
|
|
# if DEBUG_INFO: print("Write at", hex(address), size, hex(value))
|
|
# # if value == 0x98:
|
|
# # mu.mem_write(0x0, open("cfi.bin", "rb").read())
|
|
|
|
# # elif value == 0xf0:
|
|
# # mu.mem_write(0x0, open("RIFF_Nor_DCC_Test.bin", "rb").read())
|
|
|
|
# # else:
|
|
# # mu.mem_write(address, value.to_bytes(size, "little"))
|
|
|
|
# def on_error(mu, access, address, size, value, data):
|
|
# if DEBUG:
|
|
# print("Error at", hex(address), size, hex(value), "in", hex(mu.reg_read(UC_ARM_REG_PC)), "lr", hex(mu.reg_read(UC_ARM_REG_LR)))
|
|
|
|
# mu.hook_add(UC_HOOK_MEM_READ, on_read)
|
|
# mu.hook_add(UC_HOOK_MEM_WRITE, on_write)
|
|
# mu.hook_add(UC_HOOK_MEM_INVALID, on_error)
|
|
|
|
# # emulate machine code in infinite time
|
|
# mu.emu_start(0x14000000, 0x1440000)
|
|
|
|
# # now print out some registers
|
|
# print(">>> Emulation done. Below is the CPU context")
|
|
|
|
# r0 = mu.reg_read(UC_ARM_REG_R0)
|
|
# r1 = mu.reg_read(UC_ARM_REG_R1)
|
|
# print(">>> R0 = 0x%x" %r0)
|
|
# print(">>> R1 = 0x%x" %r1)
|
|
|
|
# except UcError as e:
|
|
# print("ERROR: %s" % e)
|
|
|
|
# def _dcc_read_host():
|
|
# global status_reg
|
|
# if (status_reg & 2) == 0: return 0
|
|
|
|
# print("DBG READ")
|
|
# temp = wr_reg
|
|
|
|
# status_reg &= ~2 # Debugger finally processed the data and set the W bit to low.
|
|
# return temp
|
|
|
|
# def _dcc_write_host(data):
|
|
# import time
|
|
# global status_reg, rd_reg
|
|
|
|
# while _dcc_read_status_host() & 1: time.sleep(0.1)
|
|
|
|
# print("DBG WRTIE")
|
|
# rd_reg = data
|
|
# status_reg |= 1 # With the R bit set to high, the host was as motivated to process the data.
|
|
|
|
# def _dcc_read_status_host():
|
|
# return status_reg
|
|
|
|
def _dcc_loader_read():
|
|
while (_dcc_read_status_host() & 2) == 0: time.sleep(0.1)
|
|
iCount = _dcc_read_host()
|
|
print("C:", hex(iCount))
|
|
crc = crcmod.mkCrcFun(0x104c11db7, 0xffffffff, False, 0)
|
|
hashData = bytearray()
|
|
|
|
for _ in range(iCount):
|
|
while (_dcc_read_status_host() & 2) == 0: time.sleep(0.1)
|
|
dccRead = _dcc_read_host()
|
|
print("H:", hex(dccRead))
|
|
hashData += dccRead.to_bytes(4, "little")
|
|
|
|
while (_dcc_read_status_host() & 2) == 0: time.sleep(0.1)
|
|
sum = _dcc_read_host()
|
|
|
|
hash = crc(hashData)
|
|
assert sum == hash, f"Checksum is invalid! 0x{sum:08x} != 0x{hash:08x}"
|
|
|
|
if __name__ == '__main__':
|
|
import threading
|
|
import time
|
|
|
|
emu = DCCTarget(0x78000000, open("build/qsc6055_onld_test.bin", "rb").read(), 0x20000)
|
|
|
|
t = threading.Thread(target=emu.run, daemon=True)
|
|
t.start()
|
|
|
|
#_dcc_loader_read()
|
|
print("RUN")
|
|
|
|
while True:
|
|
pass
|
|
|
|
offs = 0
|
|
|
|
if False:
|
|
while offs < 0x01000000:
|
|
_dcc_write_host(0x152 | 0x00000000)
|
|
_dcc_write_host(offs)
|
|
_dcc_write_host(0x00020000)
|
|
|
|
_dcc_loader_read()
|
|
offs += 0x20000
|
|
raise Exception("continue")
|
|
|
|
# if True:
|
|
# _dcc_write_host(0x252 | 0x00000000)
|
|
# _dcc_write_host(0x00120000)
|
|
# _dcc_write_host(0x00000080)
|
|
|
|
# _dcc_loader_read()
|
|
|
|
time.sleep(4)
|
|
|
|
print("end testing")
|
|
|