from cvalue import *
from lazytarget import *
from configuration import *
+from utils import *
import caching
import lldb
elt = elt.__getattr__(field_name)
#end of while loop
-def IterateListEntry(element, element_type, field_name):
+def IterateListEntry(element, element_type, field_name, list_prefix=''):
""" iterate over a list as defined with LIST_HEAD in bsd/sys/queue.h
params:
element - value : Value object for lh_first
element_type - str : Type of the next element
field_name - str : Name of the field in next element's structure
+ list_prefix - str : use 's' here to iterate SLIST_HEAD instead
returns:
A generator does not return. It is used for iterating
value : an object thats of type (element_type) head->le_next. Always a pointer object
for pp in IterateListEntry(headp, 'struct proc *', 'p_sibling'):
print GetProcInfo(pp)
"""
- elt = element.lh_first
+ elt = element.__getattr__(list_prefix + 'lh_first')
if type(element_type) == str:
element_type = gettype(element_type)
while unsigned(elt) != 0:
yield elt
- next_el = elt.__getattr__(field_name).le_next
+ next_el = elt.__getattr__(field_name).__getattr__(list_prefix + 'le_next')
elt = cast(next_el, element_type)
def IterateLinkageChain(queue_head, element_type, field_name, field_ofst=0):
link = link.next
-def IterateQueue(queue_head, element_ptr_type, element_field_name, backwards=False):
+def IterateQueue(queue_head, element_ptr_type, element_field_name, backwards=False, unpack_ptr_fn=None):
""" Iterate over an Element Chain queue in kernel of type queue_head_t. (osfmk/kern/queue.h method 2)
params:
queue_head - value : Value object for queue_head.
- str : OR a string describing the type. ex. 'task *'
element_field_name - str : name of the field in target struct.
backwards - backwards : traverse the queue backwards
+ unpack_ptr_fn - function : a function ptr of signature def unpack_ptr(long v) which returns long.
returns:
A generator does not return. It is used for iterating.
value : an object thats of type (element_type) queue_head->next. Always a pointer object
queue_head_addr = queue_head.GetValueAsUnsigned()
else:
queue_head_addr = queue_head.GetAddress().GetLoadAddress(LazyTarget.GetTarget())
+
+ def unpack_ptr_and_recast(v):
+ if unpack_ptr_fn is None:
+ return v
+ v_unpacked = unpack_ptr_fn(v.GetValueAsUnsigned())
+ obj = v.CreateValueFromExpression(None,'(void *)'+str(v_unpacked))
+ obj.Cast(element_ptr_type)
+ return obj
+
if backwards:
- cur_elt = queue_head.GetChildMemberWithName('prev')
+ cur_elt = unpack_ptr_and_recast(queue_head.GetChildMemberWithName('prev'))
else:
- cur_elt = queue_head.GetChildMemberWithName('next')
+ cur_elt = unpack_ptr_and_recast(queue_head.GetChildMemberWithName('next'))
while True:
elt = cur_elt.Cast(element_ptr_type)
yield value(elt)
if backwards:
- cur_elt = elt.GetChildMemberWithName(element_field_name).GetChildMemberWithName('prev')
+ cur_elt = unpack_ptr_and_recast(elt.GetChildMemberWithName(element_field_name).GetChildMemberWithName('prev'))
else:
- cur_elt = elt.GetChildMemberWithName(element_field_name).GetChildMemberWithName('next')
+ cur_elt = unpack_ptr_and_recast(elt.GetChildMemberWithName(element_field_name).GetChildMemberWithName('next'))
+
+
+def IterateRBTreeEntry(element, element_type, field_name):
+ """ iterate over a rbtree as defined with RB_HEAD in libkern/tree.h
+ element - value : Value object for rbh_root
+ element_type - str : Type of the link element
+ field_name - str : Name of the field in link element's structure
+ returns:
+ A generator does not return. It is used for iterating
+ value : an object thats of type (element_type) head->sle_next. Always a pointer object
+ """
+ elt = element.__getattr__('rbh_root')
+ if type(element_type) == str:
+ element_type = gettype(element_type)
+
+ # Walk to find min
+ parent = elt
+ while unsigned(elt) != 0:
+ parent = elt
+ elt = cast(elt.__getattr__(field_name).__getattr__('rbe_left'), element_type)
+ elt = parent
+
+ # Now elt is min
+ while unsigned(elt) != 0:
+ yield elt
+ # implementation cribbed from RB_NEXT in libkern/tree.h
+ right = cast(elt.__getattr__(field_name).__getattr__('rbe_right'), element_type)
+ if unsigned(right) != 0:
+ elt = right
+ left = cast(elt.__getattr__(field_name).__getattr__('rbe_left'), element_type)
+ while unsigned(left) != 0:
+ elt = left
+ left = cast(elt.__getattr__(field_name).__getattr__('rbe_left'), element_type)
+ else:
+
+ # avoid using GetValueFromAddress
+ addr = elt.__getattr__(field_name).__getattr__('rbe_parent')&~1
+ parent = value(elt.GetSBValue().CreateValueFromExpression(None,'(void *)'+str(addr)))
+ parent = cast(parent, element_type)
+
+ if unsigned(parent) != 0:
+ left = cast(parent.__getattr__(field_name).__getattr__('rbe_left'), element_type)
+ if (unsigned(parent) != 0) and (unsigned(elt) == unsigned(left)):
+ elt = parent
+ else:
+ if unsigned(parent) != 0:
+ right = cast(parent.__getattr__(field_name).__getattr__('rbe_right'), element_type)
+ while unsigned(parent) != 0 and (unsigned(elt) == unsigned(right)):
+ elt = parent
+
+ # avoid using GetValueFromAddress
+ addr = elt.__getattr__(field_name).__getattr__('rbe_parent')&~1
+ parent = value(elt.GetSBValue().CreateValueFromExpression(None,'(void *)'+str(addr)))
+ parent = cast(parent, element_type)
+
+ right = cast(parent.__getattr__(field_name).__getattr__('rbe_right'), element_type)
+
+ # avoid using GetValueFromAddress
+ addr = elt.__getattr__(field_name).__getattr__('rbe_parent')&~1
+ elt = value(elt.GetSBValue().CreateValueFromExpression(None,'(void *)'+str(addr)))
+ elt = cast(elt, element_type)
+
+
+def IteratePriorityQueue(root, element_type, field_name):
+ """ iterate over a priority queue as defined with struct priority_queue from osfmk/kern/priority_queue.h
+ root - value : Value object for the priority queue
+ element_type - str : Type of the link element
+ field_name - str : Name of the field in link element's structure
+ returns:
+ A generator does not return. It is used for iterating
+ value : an object thats of type (element_type). Always a pointer object
+ """
+ def _make_pqe(addr):
+ return value(root.GetSBValue().CreateValueFromExpression(None,'(struct priority_queue_entry *)'+str(addr)))
+
+ queue = [unsigned(root.pq_root_packed) & ~3]
+
+ while len(queue):
+ elt = _make_pqe(queue.pop())
+
+ while elt:
+ yield containerof(elt, element_type, field_name)
+ addr = unsigned(elt.child)
+ if addr: queue.append(addr)
+ elt = elt.next
+
+def IterateMPSCQueue(root, element_type, field_name):
+ """ iterate over an MPSC queue as defined with struct mpsc_queue_head from osfmk/kern/mpsc_queue.h
+ root - value : Value object for the mpsc queue
+ element_type - str : Type of the link element
+ field_name - str : Name of the field in link element's structure
+ returns:
+ A generator does not return. It is used for iterating
+ value : an object thats of type (element_type). Always a pointer object
+ """
+ elt = root.mpqh_head.mpqc_next
+ while unsigned(elt):
+ yield containerof(elt, element_type, field_name)
+ elt = elt.mpqc_next
class KernelTarget(object):
""" A common kernel object that provides access to kernel objects and information.
self._threads_list = []
self._tasks_list = []
self._coalitions_list = []
+ self._thread_groups = []
self._allproc = []
self._terminated_tasks_list = []
self._zones_list = []
addr = int(addr, 16)
else:
addr = int(addr)
+ addr = self.StripKernelPAC(addr)
ret_array = []
symbolicator = self._GetSymbolicator()
syms = symbolicator.symbolicate(addr)
val = ((addr + size) & (unsigned(self.GetGlobalVariable("page_size"))-1))
return (val < size and val > 0)
+ def StripUserPAC(self, addr):
+ if self.arch != 'arm64e':
+ return addr
+ T0Sz = self.GetGlobalVariable('gT0Sz')
+ return StripPAC(addr, T0Sz)
+
+ def StripKernelPAC(self, addr):
+ if self.arch != 'arm64e':
+ return addr
+ T1Sz = self.GetGlobalVariable('gT1Sz')
+ return StripPAC(addr, T1Sz)
+
+ def PhysToKVARM64(self, addr):
+ ptov_table = self.GetGlobalVariable('ptov_table')
+ for i in range(0, self.GetGlobalVariable('ptov_index')):
+ if (addr >= long(unsigned(ptov_table[i].pa))) and (addr < (long(unsigned(ptov_table[i].pa)) + long(unsigned(ptov_table[i].len)))):
+ return (addr - long(unsigned(ptov_table[i].pa)) + long(unsigned(ptov_table[i].va)))
+ return (addr - unsigned(self.GetGlobalVariable("gPhysBase")) + unsigned(self.GetGlobalVariable("gVirtBase")))
+
def PhysToKernelVirt(self, addr):
if self.arch == 'x86_64':
return (addr + unsigned(self.GetGlobalVariable('physmap_base')))
- elif self.arch == 'arm' or self.arch == 'arm64':
+ elif self.arch.startswith('arm64'):
+ return self.PhysToKVARM64(addr)
+ elif self.arch.startswith('arm'):
return (addr - unsigned(self.GetGlobalVariable("gPhysBase")) + unsigned(self.GetGlobalVariable("gVirtBase")))
else:
- raise ValueError("PhysToVirt does not support {0}".format(arch))
+ raise ValueError("PhysToVirt does not support {0}".format(self.arch))
def GetNanotimeFromAbstime(self, abstime):
""" convert absolute time (which is in MATUs) to nano seconds.
if name == 'zones' :
self._zones_list = caching.GetDynamicCacheData("kern._zones_list", [])
if len(self._zones_list) > 0: return self._zones_list
- first_zone = self.GetGlobalVariable('first_zone')
- for z in IterateLinkedList(first_zone, 'next_zone'):
- self._zones_list.append(z)
+ zone_array = self.GetGlobalVariable('zone_array')
+ for i in range(0, self.GetGlobalVariable('num_zones')):
+ self._zones_list.append(addressof(zone_array[i]))
caching.SaveDynamicCacheData("kern._zones_list", self._zones_list)
return self._zones_list
caching.SaveDynamicCacheData("kern._coalitions_list", self._coalitions_list)
return self._coalitions_list
+ if name == 'thread_groups' :
+ self._thread_groups_list = caching.GetDynamicCacheData("kern._thread_groups_list", [])
+ if len(self._thread_groups_list) > 0 : return self._thread_groups_list
+ thread_groups_queue_head = self.GetGlobalVariable('tg_queue')
+ thread_group_type = LazyTarget.GetTarget().FindFirstType('thread_group')
+ thread_groups_ptr_type = thread_group_type.GetPointerType()
+ for coal in IterateLinkageChain(addressof(thread_groups_queue_head), thread_groups_ptr_type, 'tg_queue_chain'):
+ self._thread_groups_list.append(coal)
+ caching.SaveDynamicCacheData("kern._thread_groups_list", self._thread_groups_list)
+ return self._thread_groups_list
+
if name == 'terminated_tasks' :
self._terminated_tasks_list = caching.GetDynamicCacheData("kern._terminated_tasks_list", [])
if len(self._terminated_tasks_list) > 0 : return self._terminated_tasks_list
self._ptrsize = caching.GetStaticCacheData("kern.ptrsize", None)
if self._ptrsize != None : return self._ptrsize
arch = LazyTarget.GetTarget().triple.split('-')[0]
- if arch in ('x86_64', 'arm64'):
+ if arch == 'x86_64' or arch.startswith('arm64'):
self._ptrsize = 8
else:
self._ptrsize = 4
caching.SaveStaticCacheData("kern.ptrsize", self._ptrsize)
return self._ptrsize
- return object.__getattribute__(self, name)
+ if name == 'VM_MIN_KERNEL_ADDRESS':
+ if self.arch == 'x86_64':
+ return unsigned(0xFFFFFF8000000000)
+ elif self.arch.startswith('arm64'):
+ return unsigned(0xffffffe000000000)
+ else:
+ return unsigned(0x80000000)
+ if name == 'VM_MIN_KERNEL_AND_KEXT_ADDRESS':
+ if self.arch == 'x86_64':
+ return self.VM_MIN_KERNEL_ADDRESS - 0x80000000
+ else:
+ return self.VM_MIN_KERNEL_ADDRESS
+
+ return object.__getattribute__(self, name)
projects / apple / xnu.git / blobdiff