elt = elt.__getattr__(field_name)
#end of while loop
-def IterateListEntry(element, element_type, field_name):
+def IterateSListEntry(element, element_type, field_name, slist_prefix=''):
+ """ iterate over a list as defined with SLIST_HEAD in bsd/sys/queue.h
+ params:
+ element - value : Value object for slh_first
+ element_type - str : Type of the next element
+ field_name - str : Name of the field in next 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__(slist_prefix + 'slh_first')
+ if type(element_type) == str:
+ element_type = gettype(element_type)
+ while unsigned(elt) != 0:
+ yield elt
+ next_el = elt.__getattr__(field_name).__getattr__(slist_prefix + 'sle_next')
+ elt = cast(next_el, element_type)
+
+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
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):
self._threads_list = []
self._tasks_list = []
self._coalitions_list = []
+ self._thread_groups = []
self._allproc = []
self._terminated_tasks_list = []
self._zones_list = []
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('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.
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
caching.SaveStaticCacheData("kern.ptrsize", self._ptrsize)
return self._ptrsize
+ if name == 'VM_MIN_KERNEL_ADDRESS':
+ if self.arch == 'x86_64':
+ return unsigned(0xFFFFFF8000000000)
+ elif self.arch == '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)