[apple/xnu.git] / tools / lldbmacros / usertaskdebugging / userprocess.py

import logging
import target
import struct

from xnu import *
from core.operating_system import Armv8_RegisterSet, Armv7_RegisterSet, I386_RegisterSet, X86_64RegisterSet
from core import caching

""" these defines should come from an authoritative header file """
CPU_TYPE_I386 = 0x00000007
CPU_TYPE_X86_64 = 0x01000007
CPU_TYPE_ARM = 0x0000000c
CPU_TYPE_ARM64 = 0x0100000c
CPU_TYPE_ARM64_32 = 0x0200000c

def GetRegisterSetForCPU(cputype, subtype):
    if cputype == CPU_TYPE_ARM64:
        retval = Armv8_RegisterSet
    elif cputype == CPU_TYPE_ARM64_32:
        retval = Armv8_RegisterSet
    elif cputype == CPU_TYPE_ARM:
        retval = Armv7_RegisterSet
    elif cputype == CPU_TYPE_I386:
        retval = I386_RegisterSet
    elif cputype == CPU_TYPE_X86_64:
        retval = X86_64RegisterSet

    """ crash if unknown cputype """

    return retval.register_info['registers']


class UserThreadObject(object):
    """representation of userspace thread"""
    def __init__(self, thr_obj, cputype, cpusubtype, is_kern_64bit):
        super(UserThreadObject, self).__init__()
        self.thread = thr_obj
        self.registerset = GetRegisterSetForCPU(cputype, cpusubtype)
        self.thread_id = unsigned(self.thread.thread_id)
        self.is64Bit = bool(cputype & 0x01000000)

        if self.is64Bit:
            if cputype == CPU_TYPE_X86_64:
                self.reg_type = "x86_64"
                self.saved_state = Cast(self.thread.machine.iss, 'x86_saved_state_t *').uss.ss_64
            if cputype == CPU_TYPE_ARM64:
                self.reg_type = "arm64"
                self.saved_state = self.thread.machine.upcb.uss.ss_64
        else:
            if cputype == CPU_TYPE_I386:
                self.reg_type = "i386"
                self.saved_state = Cast(self.thread.machine.iss, 'x86_saved_state_t *').uss.ss_32
            if cputype == CPU_TYPE_ARM:
                self.reg_type = "arm"
                if not is_kern_64bit:
                    self.saved_state = self.thread.machine.PcbData
                else:
                    self.saved_state = self.thread.machine.contextData.ss.uss.ss_32
            if cputype == CPU_TYPE_ARM64_32:
                self.reg_type = "arm64"
                self.saved_state = self.thread.machine.upcb.uss.ss_64

        logging.debug("created thread id 0x%x of type %s, is_kern_64bit 0x%x cputype 0x%x"
                      % (self.thread_id, self.reg_type, is_kern_64bit, cputype))

    def getRegisterValueByName(self, name):
        if self.reg_type == 'arm64':
            if name in ('x0', 'x1', 'x2', 'x3', 'x4', 'x5', 'x6', 'x7', 'x8', 'x9', 'x10', 'x11', 'x12', 'x13', 'x14', 'x15', 'x16', 'x17', 'x18', 'x19', 'x20', 'x21', 'x22', 'x23', 'x24', 'x25', 'x26', 'x27', 'x28'):
                return unsigned(getattr(self.saved_state, 'x')[int(name.strip('x'))])

            return unsigned(getattr(self.saved_state, name))

        if self.reg_type == "x86_64":
            if name in ('rip', 'rflags', 'cs', 'rsp', 'cpu'):
                return unsigned(getattr(self.saved_state.isf, name))
            return unsigned(getattr(self.saved_state, name))

        if self.reg_type == "arm":
            if name in ('r0', 'r1', 'r2', 'r3', 'r4', 'r5', 'r6', 'r7', 'r8', 'r9', 'r10', 'r11', 'r12'):
                retval = unsigned(getattr(self.saved_state, 'r')[int(name.strip('r'))])
            else:
                retval = unsigned(getattr(self.saved_state, name))
            return retval

        #TODO for i386

    def getName(self):
        return str(self.thread_id)

    def getRegisterData(self, reg_num):
        """ returns None if there is error """
        if reg_num < 0 or reg_num >= len(self.registerset):
            logging.warning("regnum %d is not defined for thread_id 0x%x" % (reg_num, self.thread_id))
            return None
        return self.getRegisterValueByName(self.registerset[reg_num]['name'])


class UserProcess(target.Process):
    """ Represent a user process and thread states """
    def __init__(self, task):
        self.task = task
        self.proc = Cast(task.bsd_info, 'proc_t')
        dataregisters64bit = False
        ptrsize = 4

        if task.t_flags & 0x1:
            ptrsize = 8
        if task.t_flags & 0x2:
            dataregisters64bit = True

        is_kern_64bit = kern.arch in ['x86_64', 'x86_64h', 'arm64', 'arm64e']

        self.cputype = unsigned(self.proc.p_cputype)
        self.cpusubtype = unsigned(self.proc.p_cpusubtype)

        super(UserProcess, self).__init__(self.cputype, self.cpusubtype, ptrsize)
        dbg_message = "process:%s is64bit:%d ptrsize:%d cputype:0x%x cpusubtype:0x%x" % (hex(self.proc), int(dataregisters64bit), ptrsize, self.cputype, self.cpusubtype)
        self.proc_platform = int(self.proc.p_platform)
        if self.proc_platform == xnudefines.P_PLATFORM_MACOS:
            self.hinfo['ostype'] = 'macosx'
        elif self.proc_platform == xnudefines.P_PLATFORM_WATCHOS:
            self.hinfo['ostype'] = 'watchos'
        elif self.proc_platform == xnudefines.P_PLATFORM_TVOS:
            self.hinfo['ostype'] = 'tvos'
        else:
            self.hinfo['ostype'] = 'ios'
        dbg_message += " ostype:%s" % self.hinfo['ostype']

        if is_kern_64bit and str(kern.arch).lower().startswith('arm'):
            addressing_bits = 64 - int(kern.globals.gT1Sz)
            self.hinfo['addressing_bits'] = addressing_bits
            dbg_message += " addressing_bits:%d" % addressing_bits

        self.registerset = GetRegisterSetForCPU(self.cputype, self.cpusubtype)
        logging.info(dbg_message)
        self.threads = {}
        self.threads_ids_list = []
        logging.debug("iterating over threads in process")
        for thval in IterateQueue(task.threads, 'thread *', 'task_threads'):
            self.threads[unsigned(thval.thread_id)] = UserThreadObject(thval, self.cputype, self.cpusubtype, is_kern_64bit)
            self.threads_ids_list.append(unsigned(thval.thread_id))

    def getRegisterDataForThread(self, th_id, reg_num):
        if th_id not in self.threads:
            logging.critical("0x%x thread id is not found in this task")
            return ''
        if reg_num < 0 or reg_num >= len(self.registerset):
            logging.warning("regnum %d is not defined for thread_id 0x%x" % (reg_num, th_id))
            return ''
        value = self.threads[th_id].getRegisterData(reg_num)
        return self.encodeRegisterData(value, bytesize=self.registerset[reg_num]['bitsize']/8)

    def getRegisterCombinedDataForThread(self, th_id):
        if th_id not in self.threads:
            logging.critical("0x%x thread id is not found in this task" % th_id)
            return ''
        cur_thread = self.threads[th_id]
        retval = 'thread:%s;name:%s;' % (self.encodeThreadID(th_id), cur_thread.getName())
        pos = 0
        for rinfo in self.registerset:
            name = rinfo['name']
            format = "%02x:%s;"
            value = cur_thread.getRegisterValueByName(name)
            value_endian_correct_str = self.encodeRegisterData(value, bytesize=(rinfo['bitsize']/8))
            retval += format % (pos, value_endian_correct_str)
            pos += 1
        return retval

    def getThreadStopInfo(self, th_id):
        if th_id not in self.threads:
            logging.critical("0x%x thread id is not found in this task")
            return ''
        return 'T02' + self.getRegisterCombinedDataForThread(th_id) + 'threads:' + self.getThreadsInfo()+';'

    def getRegisterInfo(self, regnum):
        #something similar to
        #"name:x1;bitsize:64;offset:8;encoding:uint;format:hex;gcc:1;dwarf:1;set:General Purpose Registers;"
        if regnum >= len(self.registerset):
            logging.debug("No register_info for number %d." % regnum)
            return 'E45'

        rinfo = self.registerset[regnum]
        retval = ''
        for i in rinfo.keys():
            i_val = str(rinfo[i])
            if i == 'set':
                i_val = 'General Purpose Registers'
            retval += '%s:%s;' % (str(i), i_val)

        return retval

    def getProcessInfo(self):
        retval = ''
        #pid:d22c;parent-pid:d34d;real-uid:ecf;real-gid:b;effective-uid:ecf;effective-gid:b;cputype:1000007;cpusubtype:3;
        #ostype:macosx;vendor:apple;endian:little;ptrsize:8;
        pinfo = {'effective-uid': 'ecf', 'effective-gid': 'b', 'endian': 'little', 'vendor': 'apple'}
        pinfo['pid'] = "%x" % (GetProcPIDForTask(self.task))
        pinfo['parent-pid'] = "%x" % (unsigned(self.proc.p_ppid))
        pinfo['ptrsize'] = str(self.ptrsize)
        pinfo['ostype'] = 'macosx'
        pinfo['cputype'] = "%x" % self.cputype
        pinfo['cpusubtype'] = "%x" % self.cpusubtype
        pinfo['real-uid'] = "%x" % (unsigned(self.proc.p_ruid))
        pinfo['real-gid'] = "%x" % (unsigned(self.proc.p_rgid))
        if str(kern.arch).find('arm') >= 0:
            pinfo['ostype'] = 'ios'
        for i in pinfo.keys():
            i_val = str(pinfo[i])
            retval += '%s:%s;' % (str(i), i_val)
        return retval

    def readMemory(self, address, size):
        cache_key = "{}-{}-{}".format(hex(self.task), hex(address), size)
        cache_data = caching.GetDynamicCacheData(cache_key)
        if cache_data:
            return self.encodeByteString(cache_data)
        data = GetUserDataAsString(self.task, address, size)
        if not data:
            logging.error("Failed to read memory task:{: <#018x} {: <#018x} {:d}".format(self.task, address, size))
        else:
            caching.SaveDynamicCacheData(cache_key, data)
        return self.encodeByteString(data)

    def getSharedLibInfoAddress(self):
        return unsigned(self.task.all_image_info_addr)
Commit	Line	Data
39037602 A	1	import logging
	2	import target
	3	import struct
	4
	5	from xnu import *
	6	from core.operating_system import Armv8_RegisterSet, Armv7_RegisterSet, I386_RegisterSet, X86_64RegisterSet
f427ee49	7	from core import caching
39037602	8
5ba3f43e	9	""" these defines should come from an authoritative header file """
39037602 A	10	CPU_TYPE_I386 = 0x00000007
	11	CPU_TYPE_X86_64 = 0x01000007
	12	CPU_TYPE_ARM = 0x0000000c
	13	CPU_TYPE_ARM64 = 0x0100000c
cb323159	14	CPU_TYPE_ARM64_32 = 0x0200000c
39037602	15
39037602	16	def GetRegisterSetForCPU(cputype, subtype):
d9a64523	17	if cputype == CPU_TYPE_ARM64:
5ba3f43e	18	retval = Armv8_RegisterSet
cb323159 A	19	elif cputype == CPU_TYPE_ARM64_32:
cb323159 A	20	retval = Armv8_RegisterSet
5ba3f43e A	21	elif cputype == CPU_TYPE_ARM:
5ba3f43e A	22	retval = Armv7_RegisterSet
39037602 A	23	elif cputype == CPU_TYPE_I386:
39037602 A	24	retval = I386_RegisterSet
5ba3f43e A	25	elif cputype == CPU_TYPE_X86_64:
5ba3f43e A	26	retval = X86_64RegisterSet
f427ee49	27
5ba3f43e	28	""" crash if unknown cputype """
39037602 A	29
	30	return retval.register_info['registers']
	31
	32
	33	class UserThreadObject(object):
	34	"""representation of userspace thread"""
d9a64523	35	def __init__(self, thr_obj, cputype, cpusubtype, is_kern_64bit):
39037602 A	36	super(UserThreadObject, self).__init__()
	37	self.thread = thr_obj
	38	self.registerset = GetRegisterSetForCPU(cputype, cpusubtype)
	39	self.thread_id = unsigned(self.thread.thread_id)
	40	self.is64Bit = bool(cputype & 0x01000000)
39037602 A	41
	42	if self.is64Bit:
	43	if cputype == CPU_TYPE_X86_64:
	44	self.reg_type = "x86_64"
	45	self.saved_state = Cast(self.thread.machine.iss, 'x86_saved_state_t *').uss.ss_64
	46	if cputype == CPU_TYPE_ARM64:
	47	self.reg_type = "arm64"
	48	self.saved_state = self.thread.machine.upcb.uss.ss_64
	49	else:
	50	if cputype == CPU_TYPE_I386:
	51	self.reg_type = "i386"
	52	self.saved_state = Cast(self.thread.machine.iss, 'x86_saved_state_t *').uss.ss_32
	53	if cputype == CPU_TYPE_ARM:
	54	self.reg_type = "arm"
d9a64523	55	if not is_kern_64bit:
39037602 A	56	self.saved_state = self.thread.machine.PcbData
	57	else:
	58	self.saved_state = self.thread.machine.contextData.ss.uss.ss_32
cb323159 A	59	if cputype == CPU_TYPE_ARM64_32:
	60	self.reg_type = "arm64"
	61	self.saved_state = self.thread.machine.upcb.uss.ss_64
d9a64523 A	62
	63	logging.debug("created thread id 0x%x of type %s, is_kern_64bit 0x%x cputype 0x%x"
	64	% (self.thread_id, self.reg_type, is_kern_64bit, cputype))
39037602 A	65
	66	def getRegisterValueByName(self, name):
	67	if self.reg_type == 'arm64':
	68	if name in ('x0', 'x1', 'x2', 'x3', 'x4', 'x5', 'x6', 'x7', 'x8', 'x9', 'x10', 'x11', 'x12', 'x13', 'x14', 'x15', 'x16', 'x17', 'x18', 'x19', 'x20', 'x21', 'x22', 'x23', 'x24', 'x25', 'x26', 'x27', 'x28'):
	69	return unsigned(getattr(self.saved_state, 'x')[int(name.strip('x'))])
	70
	71	return unsigned(getattr(self.saved_state, name))
	72
	73	if self.reg_type == "x86_64":
	74	if name in ('rip', 'rflags', 'cs', 'rsp', 'cpu'):
	75	return unsigned(getattr(self.saved_state.isf, name))
	76	return unsigned(getattr(self.saved_state, name))
	77
	78	if self.reg_type == "arm":
	79	if name in ('r0', 'r1', 'r2', 'r3', 'r4', 'r5', 'r6', 'r7', 'r8', 'r9', 'r10', 'r11', 'r12'):
	80	retval = unsigned(getattr(self.saved_state, 'r')[int(name.strip('r'))])
	81	else:
	82	retval = unsigned(getattr(self.saved_state, name))
	83	return retval
	84
	85	#TODO for i386
	86
	87	def getName(self):
	88	return str(self.thread_id)
	89
	90	def getRegisterData(self, reg_num):
	91	""" returns None if there is error """
	92	if reg_num < 0 or reg_num >= len(self.registerset):
	93	logging.warning("regnum %d is not defined for thread_id 0x%x" % (reg_num, self.thread_id))
	94	return None
	95	return self.getRegisterValueByName(self.registerset[reg_num]['name'])
	96
	97
	98	class UserProcess(target.Process):
	99	""" Represent a user process and thread states """
	100	def __init__(self, task):
	101	self.task = task
	102	self.proc = Cast(task.bsd_info, 'proc_t')
	103	dataregisters64bit = False
	104	ptrsize = 4
	105
	106	if task.t_flags & 0x1:
	107	ptrsize = 8
	108	if task.t_flags & 0x2:
d9a64523	109	dataregisters64bit = True
39037602	110
cb323159	111	is_kern_64bit = kern.arch in ['x86_64', 'x86_64h', 'arm64', 'arm64e']
5ba3f43e	112
d9a64523 A	113	self.cputype = unsigned(self.proc.p_cputype)
d9a64523 A	114	self.cpusubtype = unsigned(self.proc.p_cpusubtype)
39037602	115
d9a64523	116	super(UserProcess, self).__init__(self.cputype, self.cpusubtype, ptrsize)
f427ee49 A	117	dbg_message = "process:%s is64bit:%d ptrsize:%d cputype:0x%x cpusubtype:0x%x" % (hex(self.proc), int(dataregisters64bit), ptrsize, self.cputype, self.cpusubtype)
	118	self.proc_platform = int(self.proc.p_platform)
	119	if self.proc_platform == xnudefines.P_PLATFORM_MACOS:
	120	self.hinfo['ostype'] = 'macosx'
	121	elif self.proc_platform == xnudefines.P_PLATFORM_WATCHOS:
	122	self.hinfo['ostype'] = 'watchos'
	123	elif self.proc_platform == xnudefines.P_PLATFORM_TVOS:
	124	self.hinfo['ostype'] = 'tvos'
	125	else:
39037602	126	self.hinfo['ostype'] = 'ios'
f427ee49 A	127	dbg_message += " ostype:%s" % self.hinfo['ostype']
	128
	129	if is_kern_64bit and str(kern.arch).lower().startswith('arm'):
	130	addressing_bits = 64 - int(kern.globals.gT1Sz)
	131	self.hinfo['addressing_bits'] = addressing_bits
	132	dbg_message += " addressing_bits:%d" % addressing_bits
39037602	133
d9a64523	134	self.registerset = GetRegisterSetForCPU(self.cputype, self.cpusubtype)
f427ee49	135	logging.info(dbg_message)
39037602 A	136	self.threads = {}
	137	self.threads_ids_list = []
	138	logging.debug("iterating over threads in process")
	139	for thval in IterateQueue(task.threads, 'thread *', 'task_threads'):
d9a64523	140	self.threads[unsigned(thval.thread_id)] = UserThreadObject(thval, self.cputype, self.cpusubtype, is_kern_64bit)
39037602 A	141	self.threads_ids_list.append(unsigned(thval.thread_id))
	142
	143	def getRegisterDataForThread(self, th_id, reg_num):
	144	if th_id not in self.threads:
	145	logging.critical("0x%x thread id is not found in this task")
	146	return ''
	147	if reg_num < 0 or reg_num >= len(self.registerset):
	148	logging.warning("regnum %d is not defined for thread_id 0x%x" % (reg_num, th_id))
	149	return ''
	150	value = self.threads[th_id].getRegisterData(reg_num)
	151	return self.encodeRegisterData(value, bytesize=self.registerset[reg_num]['bitsize']/8)
	152
	153	def getRegisterCombinedDataForThread(self, th_id):
	154	if th_id not in self.threads:
	155	logging.critical("0x%x thread id is not found in this task" % th_id)
	156	return ''
	157	cur_thread = self.threads[th_id]
	158	retval = 'thread:%s;name:%s;' % (self.encodeThreadID(th_id), cur_thread.getName())
	159	pos = 0
	160	for rinfo in self.registerset:
	161	name = rinfo['name']
	162	format = "%02x:%s;"
	163	value = cur_thread.getRegisterValueByName(name)
	164	value_endian_correct_str = self.encodeRegisterData(value, bytesize=(rinfo['bitsize']/8))
	165	retval += format % (pos, value_endian_correct_str)
	166	pos += 1
	167	return retval
	168
	169	def getThreadStopInfo(self, th_id):
	170	if th_id not in self.threads:
	171	logging.critical("0x%x thread id is not found in this task")
	172	return ''
	173	return 'T02' + self.getRegisterCombinedDataForThread(th_id) + 'threads:' + self.getThreadsInfo()+';'
	174
	175	def getRegisterInfo(self, regnum):
	176	#something similar to
	177	#"name:x1;bitsize:64;offset:8;encoding:uint;format:hex;gcc:1;dwarf:1;set:General Purpose Registers;"
f427ee49	178	if regnum >= len(self.registerset):
39037602 A	179	logging.debug("No register_info for number %d." % regnum)
	180	return 'E45'
	181
	182	rinfo = self.registerset[regnum]
	183	retval = ''
	184	for i in rinfo.keys():
	185	i_val = str(rinfo[i])
	186	if i == 'set':
	187	i_val = 'General Purpose Registers'
	188	retval += '%s:%s;' % (str(i), i_val)
	189
	190	return retval
	191
	192	def getProcessInfo(self):
	193	retval = ''
	194	#pid:d22c;parent-pid:d34d;real-uid:ecf;real-gid:b;effective-uid:ecf;effective-gid:b;cputype:1000007;cpusubtype:3;
	195	#ostype:macosx;vendor:apple;endian:little;ptrsize:8;
	196	pinfo = {'effective-uid': 'ecf', 'effective-gid': 'b', 'endian': 'little', 'vendor': 'apple'}
	197	pinfo['pid'] = "%x" % (GetProcPIDForTask(self.task))
	198	pinfo['parent-pid'] = "%x" % (unsigned(self.proc.p_ppid))
	199	pinfo['ptrsize'] = str(self.ptrsize)
	200	pinfo['ostype'] = 'macosx'
	201	pinfo['cputype'] = "%x" % self.cputype
	202	pinfo['cpusubtype'] = "%x" % self.cpusubtype
	203	pinfo['real-uid'] = "%x" % (unsigned(self.proc.p_ruid))
	204	pinfo['real-gid'] = "%x" % (unsigned(self.proc.p_rgid))
	205	if str(kern.arch).find('arm') >= 0:
	206	pinfo['ostype'] = 'ios'
	207	for i in pinfo.keys():
	208	i_val = str(pinfo[i])
	209	retval += '%s:%s;' % (str(i), i_val)
	210	return retval
	211
	212	def readMemory(self, address, size):
f427ee49 A	213	cache_key = "{}-{}-{}".format(hex(self.task), hex(address), size)
	214	cache_data = caching.GetDynamicCacheData(cache_key)
	215	if cache_data:
	216	return self.encodeByteString(cache_data)
39037602 A	217	data = GetUserDataAsString(self.task, address, size)
	218	if not data:
	219	logging.error("Failed to read memory task:{: <#018x} {: <#018x} {:d}".format(self.task, address, size))
f427ee49 A	220	else:
f427ee49 A	221	caching.SaveDynamicCacheData(cache_key, data)
39037602 A	222	return self.encodeByteString(data)
	223
	224	def getSharedLibInfoAddress(self):
	225	return unsigned(self.task.all_image_info_addr)