from xnu import *
from utils import *
from kdp import *
+from core import caching
import sys
+import lldb
+from collections import deque
######################################
# Globals
v.GetSBValue().SetPreferDynamicValue(lldb.eNoDynamicValues)
return v
+#####################################
+# Classes.
+#####################################
+class PreoslogHeader(object):
+ """
+ Represents preoslog buffer header. There's no symbol in the kernel for it.
+ """
+ valid_magic = "POSL"
+ def __init__(self):
+ self.magic = ""
+ self.offset = 0
+ self.size = 0
+ self.source = 0
+ self.wrapped = 0
+ self.data = None
+
######################################
# Type Summaries
######################################
print out_string
+@lldb_type_summary(['IOPMWorkQueue *'])
+@header("")
+def GetIOPMWorkQueueSummary(wq):
+ out_str = ""
+ ioservicepm_header = "{:<20s}{:<4s}{:<4s}{:<4s}{:<4s}\n"
+ iopmrequest_indent = " "
+ iopmrequest_header = iopmrequest_indent + "{:<20s}{:<6s}{:<20s}{:<20s}{:<12s}{:<12s}{:<20s}{:<20s}{:<20s}\n"
+
+ for next in IterateQueue(wq.fWorkQueue, 'IOServicePM *', 'WorkChain'):
+ out_str += ioservicepm_header.format("IOService", "ps", "ms", "wr", "name")
+ out_str += "0x{:<16x} {:<2d} {:<2d} {:<2d} {:<s}\n".format(
+ next.Owner, next.CurrentPowerState, next.MachineState, next.WaitReason, next.Name)
+ out_str += iopmrequest_header.format("IOPMRequest", "type", "next_req", "root_req", "work_wait", "free_wait", "arg0", "arg1", "arg2")
+ for request in IterateQueue(next.RequestHead, 'IOPMRequest *', 'fCommandChain'):
+ out_str += iopmrequest_indent
+ out_str += "0x{:<16x} 0x{:<2x} 0x{:<16x} 0x{:<16x}".format(
+ request, request.fRequestType, request.fRequestNext, request.fRequestRoot)
+ out_str += " 0x{:<8x} 0x{:<8x}".format(
+ request.fWorkWaitCount, request.fFreeWaitCount)
+ out_str += " 0x{:<16x} 0x{:<16x} 0x{:<16x}\n".format(
+ request.fArg0, request.fArg1, request.fArg2)
+ return out_str
+
+@lldb_command('showiopmqueues')
+def ShowIOPMQueues(cmd_args=None):
+ """ Show IOKit power management queues and IOPMRequest objects.
+ """
+ print "IOPMWorkQueue 0x{:<16x} ({:<d} IOServicePM)\n".format(
+ kern.globals.gIOPMWorkQueue, kern.globals.gIOPMWorkQueue.fQueueLength)
+ print GetIOPMWorkQueueSummary(kern.globals.gIOPMWorkQueue)
+
+@lldb_type_summary(['IOService *'])
+@header("")
+def GetIOPMInterest(service):
+ iopm = CastIOKitClass(service.pwrMgt, 'IOServicePM *')
+ if unsigned(iopm) == 0:
+ print("error: no IOServicePM")
+ return
+
+ list = CastIOKitClass(iopm.InterestedDrivers, 'IOPMinformeeList *')
+ out_str = "IOServicePM 0x{:<16x} ({:<d} interest, {:<d} pending ack)\n".format(
+ iopm, list.length, iopm.HeadNotePendingAcks)
+ if list.length == 0:
+ return
+
+ out_str += " {:<20s}{:<8s}{:<10s}{:<20s}{:<20s}{:<20s}{:<s}\n".format(
+ "informee", "active", "ticks", "notifyTime", "service", "regId", "name")
+ next = CastIOKitClass(list.firstItem, 'IOPMinformee *')
+ while unsigned(next) != 0:
+ driver = CastIOKitClass(next.whatObject, 'IOService *')
+ name = GetRegistryEntryName(driver)
+ reg_id = CastIOKitClass(driver, 'IORegistryEntry *').reserved.fRegistryEntryID;
+ out_str += " 0x{:<16x} {:<6s} {:<8d} 0x{:<16x} 0x{:<16x} 0x{:<16x} {:<s}\n".format(
+ next, "Yes" if next.active != 0 else "No" , next.timer, next.startTime, next.whatObject, reg_id, name)
+ next = CastIOKitClass(next.nextInList, 'IOPMinformee *')
+ return out_str
+
+@lldb_command('showiopminterest')
+def ShowIOPMInterest(cmd_args=None):
+ """ Show the interested drivers for an IOService.
+ syntax: (lldb) showiopminterest <IOService>
+ """
+ if not cmd_args:
+ print "Please specify the address of the IOService"
+ print ShowIOPMInterest.__doc__
+ return
+
+ obj = kern.GetValueFromAddress(cmd_args[0], 'IOService *')
+ print GetIOPMInterest(obj)
+
+@lldb_command("showinterruptvectors")
+def ShowInterruptVectorInfo(cmd_args=None):
+ """
+ Shows interrupt vectors.
+ """
+
+ # Constants
+ kInterruptTriggerModeMask = 0x01
+ kInterruptTriggerModeEdge = 0x00
+ kInterruptTriggerModeLevel = kInterruptTriggerModeMask
+ kInterruptPolarityMask = 0x02
+ kInterruptPolarityHigh = 0x00
+ kInterruptPolarityLow = kInterruptPolarityMask
+ kInterruptShareableMask = 0x04
+ kInterruptNotShareable = 0x00
+ kInterruptIsShareable = kInterruptShareableMask
+ kIOInterruptTypePCIMessaged = 0x00010000
+
+ # Get all interrupt controllers
+ interrupt_controllers = list(SearchInterruptControllerDrivers())
+
+ print("Interrupt controllers: ")
+ for ic in interrupt_controllers:
+ print(" {}".format(ic))
+ print("")
+
+ # Iterate over all entries in the registry
+ for entry in GetMatchingEntries(lambda _: True):
+ # Get the name of the entry
+ entry_name = GetRegistryEntryName(entry)
+
+ # Get the location of the entry
+ entry_location = GetRegistryEntryLocationInPlane(entry, kern.globals.gIOServicePlane)
+ if entry_location is None:
+ entry_location = ""
+ else:
+ entry_location = "@" + entry_location
+
+ # Get the interrupt properties
+ (msi_mode, vectorDataList, vectorContList) = GetRegistryEntryInterruptProperties(entry)
+ should_print = False
+ out_str = ""
+ for (vector_data, vector_cont) in zip(vectorDataList, vectorContList):
+ # vector_cont is the name of the interrupt controller. Find the matching controller from
+ # the list of controllers obtained earlier
+ matching_ics = filter(lambda ic: ic.name == vector_cont, interrupt_controllers)
+
+ if len(matching_ics) > 0:
+ should_print = True
+ # Take the first match
+ matchingIC = matching_ics[0]
+
+ # Use the vector_data to determine the vector and any flags
+ data_ptr = vector_data.data
+ data_length = vector_data.length
+
+ # Dereference vector_data as a uint32_t * and add the base vector number
+ gsi = unsigned(dereference(Cast(data_ptr, 'uint32_t *')))
+ gsi += matchingIC.base_vector_number
+
+ # If data_length is >= 8 then vector_data contains interrupt flags
+ if data_length >= 8:
+ # Add sizeof(uint32_t) to data_ptr to get the flags pointer
+ flags_ptr = kern.GetValueFromAddress(unsigned(data_ptr) + sizeof("uint32_t"))
+ flags = unsigned(dereference(Cast(flags_ptr, 'uint32_t *')))
+ out_str += " +----- [Interrupt Controller {ic}] vector {gsi}, {trigger_level}, {active}, {shareable}{messaged}\n" \
+ .format(ic=matchingIC.name, gsi=hex(gsi),
+ trigger_level="level trigger" if flags & kInterruptTriggerModeLevel else "edge trigger",
+ active="active low" if flags & kInterruptPolarityLow else "active high",
+ shareable="shareable" if flags & kInterruptIsShareable else "exclusive",
+ messaged=", messaged" if flags & kIOInterruptTypePCIMessaged else "")
+ else:
+ out_str += " +----- [Interrupt Controller {ic}] vector {gsi}\n".format(ic=matchingIC.name, gsi=hex(gsi))
+ if should_print:
+ print("[ {entry_name}{entry_location} ]{msi_mode}\n{out_str}" \
+ .format(entry_name=entry_name,
+ entry_location=entry_location,
+ msi_mode=" - MSIs enabled" if msi_mode else "",
+ out_str=out_str))
+
+@lldb_command("showiokitclasshierarchy")
+def ShowIOKitClassHierarchy(cmd_args=None):
+ """
+ Show class hierarchy for a IOKit class
+ """
+ if not cmd_args:
+ print("Usage: showiokitclasshierarchy <IOKit class name>")
+ return
+
+ class_name = cmd_args[0]
+ metaclasses = GetMetaClasses()
+ if class_name not in metaclasses:
+ print("Class {} does not exist".format(class_name))
+ return
+ metaclass = metaclasses[class_name]
+
+ # loop over superclasses
+ hierarchy = []
+ current_metaclass = metaclass
+ while current_metaclass is not None:
+ hierarchy.insert(0, current_metaclass)
+ current_metaclass = current_metaclass.superclass()
+
+ for (index, mc) in enumerate(hierarchy):
+ indent = (" " * index) + "+---"
+ print("{}[ {} ] {}".format(indent, str(mc.className()), str(mc.data())))
+
+
######################################
# Helper routines
######################################
return registry_object
return None
-def LookupKeyInOSDict(osdict, key):
+def CompareStringToOSSymbol(string, os_sym):
+ """
+ Lexicographically compare python string to OSSymbol
+ Params:
+ string - python string
+ os_sym - OSSymbol
+
+ Returns:
+ 0 if string == os_sym
+ 1 if string > os_sym
+ -1 if string < os_sym
+ """
+ os_sym_str = GetString(os_sym)
+ if string > os_sym_str:
+ return 1
+ elif string < os_sym_str:
+ return -1
+ else:
+ return 0
+
+class IOKitMetaClass(object):
+ """
+ A class that represents a IOKit metaclass. This is used to represent the
+ IOKit inheritance hierarchy.
+ """
+
+ def __init__(self, meta):
+ """
+ Initialize a IOKitMetaClass object.
+
+ Args:
+ meta (core.cvalue.value): A LLDB value representing a
+ OSMetaClass *.
+ """
+ self._meta = meta
+ self._superclass = None
+
+ def data(self):
+ return self._meta
+
+ def setSuperclass(self, superclass):
+ """
+ Set the superclass for this metaclass.
+
+ Args:
+ superclass (core.cvalue.value): A LLDB value representing a
+ OSMetaClass *.
+ """
+ self._superclass = superclass
+
+ def superclass(self):
+ """
+ Get the superclass for this metaclass (set by the setSuperclass method).
+
+ Returns:
+ core.cvalue.value: A LLDB value representing a OSMetaClass *.
+ """
+ return self._superclass
+
+ def className(self):
+ """
+ Get the name of the class this metaclass represents.
+
+ Returns:
+ str: The class name
+ """
+ return self._meta.className.string
+
+ def inheritsFrom(self, other):
+ """
+ Check if the class represented by this metaclass inherits from a class
+ represented by another metaclass.
+
+ Args:
+ other (IOKitMetaClass): The other metaclass
+
+ Returns:
+ bool: Returns True if this class inherits from the other class and
+ False otherwise.
+ """
+ current = self
+ while current is not None:
+ if current == other:
+ return True
+ else:
+ current = current.superclass()
+
+
+def GetRegistryEntryClassName(entry):
+ """
+ Get the class name of a registry entry.
+
+ Args:
+ entry (core.cvalue.value): A LLDB value representing a
+ IORegistryEntry *.
+
+ Returns:
+ str: The class name of the entry or None if a class name could not be
+ found.
+ """
+ # Check using IOClass key
+ result = LookupKeyInOSDict(entry.fPropertyTable, kern.globals.gIOClassKey)
+ if result is not None:
+ return GetString(result).replace("\"", "")
+ else:
+ # Use the vtable of the entry to determine the concrete type
+ vt = dereference(Cast(entry, 'uintptr_t *')) - 2 * sizeof('uintptr_t')
+ vt = kern.StripKernelPAC(vt)
+ vtype = kern.SymbolicateFromAddress(vt)
+ if len(vtype) > 0:
+ vtableName = vtype[0].GetName()
+ return vtableName[11:] # strip off "vtable for "
+ else:
+ return None
+
+
+def GetRegistryEntryName(entry):
+ """
+ Get the name of a registry entry.
+
+ Args:
+ entry (core.cvalue.value): A LLDB value representing a
+ IORegistryEntry *.
+
+ Returns:
+ str: The name of the entry or None if a name could not be found.
+ """
+ name = None
+
+ # First check the IOService plane nameKey
+ result = LookupKeyInOSDict(entry.fRegistryTable, kern.globals.gIOServicePlane.nameKey)
+ if result is not None:
+ name = GetString(result)
+
+ # Check the global IOName key
+ if name is None:
+ result = LookupKeyInOSDict(entry.fRegistryTable, kern.globals.gIONameKey)
+ if result is not None:
+ name = GetString(result)
+
+ # Check the IOClass key
+ if name is None:
+ result = LookupKeyInOSDict(entry.fPropertyTable, kern.globals.gIOClassKey)
+ if result is not None:
+ name = GetString(result)
+
+ # Remove extra quotes
+ if name is not None:
+ return name.replace("\"", "")
+ else:
+ return GetRegistryEntryClassName(entry)
+
+
+def GetRegistryEntryLocationInPlane(entry, plane):
+ """
+ Get the registry entry location in a IOKit plane.
+
+ Args:
+ entry (core.cvalue.value): A LLDB value representing a
+ IORegistryEntry *.
+ plane: An IOKit plane such as kern.globals.gIOServicePlane.
+
+ Returns:
+ str: The location of the entry or None if a location could not be
+ found.
+ """
+ # Check the plane's pathLocationKey
+ sym = LookupKeyInOSDict(entry.fRegistryTable, plane.pathLocationKey)
+
+ # Check the global IOLocation key
+ if sym is None:
+ sym = LookupKeyInOSDict(entry.fRegistryTable, kern.globals.gIOLocationKey)
+ if sym is not None:
+ return GetString(sym).replace("\"", "")
+ else:
+ return None
+
+
+def GetMetaClasses():
+ """
+ Enumerate all IOKit metaclasses. Uses dynamic caching.
+
+ Returns:
+ Dict[str, IOKitMetaClass]: A dictionary mapping each metaclass name to
+ a IOKitMetaClass object representing the metaclass.
+ """
+ METACLASS_CACHE_KEY = "iokit_metaclasses"
+ cached_data = caching.GetDynamicCacheData(METACLASS_CACHE_KEY)
+
+ # If we have cached data, return immediately
+ if cached_data is not None:
+ return cached_data
+
+ # This method takes a while, so it prints a progress indicator
+ print("Enumerating IOKit metaclasses: ")
+
+ # Iterate over all classes present in sAllClassesDict
+ idx = 0
+ count = unsigned(kern.globals.sAllClassesDict.count)
+ metaclasses_by_address = {}
+ while idx < count:
+ # Print progress after every 10 items
+ if idx % 10 == 0:
+ print(" {} metaclass structures parsed...".format(idx))
+
+ # Address of metaclass
+ address = kern.globals.sAllClassesDict.dictionary[idx].value
+
+ # Create IOKitMetaClass and store in dict
+ metaclasses_by_address[int(address)] = IOKitMetaClass(CastIOKitClass(kern.globals.sAllClassesDict.dictionary[idx].value, 'OSMetaClass *'))
+ idx += 1
+
+ print(" Enumerated {} metaclasses.".format(count))
+
+ # At this point, each metaclass is independent of each other. We don't have superclass links set up yet.
+
+ for (address, metaclass) in metaclasses_by_address.items():
+ # Get the address of the superclass using the superClassLink in IOMetaClass
+ superclass_address = int(metaclass.data().superClassLink)
+
+ # Skip null superclass
+ if superclass_address == 0:
+ continue
+
+ # Find the superclass object in the dict
+ if superclass_address in metaclasses_by_address:
+ metaclass.setSuperclass(metaclasses_by_address[superclass_address])
+ else:
+ print("warning: could not find superclass for {}".format(str(metaclass.data())))
+
+ # This method returns a dictionary mapping each class name to the associated metaclass object
+ metaclasses_by_name = {}
+ for (_, metaclass) in metaclasses_by_address.items():
+ metaclasses_by_name[str(metaclass.className())] = metaclass
+
+ # Save the result in the cache
+ caching.SaveDynamicCacheData(METACLASS_CACHE_KEY, metaclasses_by_name)
+
+ return metaclasses_by_name
+
+
+def GetMatchingEntries(matcher):
+ """
+ Iterate over the IOKit registry and find entries that match specific
+ criteria.
+
+ Args:
+ matcher (function): A matching function that returns True for a match
+ and False otherwise.
+
+ Yields:
+ core.cvalue.value: LLDB values that represent IORegistryEntry * for
+ each registry entry found.
+ """
+
+ # Perform a BFS over the IOKit registry tree
+ bfs_queue = deque()
+ bfs_queue.append(kern.globals.gRegistryRoot)
+ while len(bfs_queue) > 0:
+ # Dequeue an entry
+ entry = bfs_queue.popleft()
+
+ # Check if entry matches
+ if matcher(entry):
+ yield entry
+
+ # Find children of this entry and enqueue them
+ child_array = LookupKeyInOSDict(entry.fRegistryTable, kern.globals.gIOServicePlane.keys[1])
+ if child_array is not None:
+ idx = 0
+ ca = CastIOKitClass(child_array, 'OSArray *')
+ count = unsigned(ca.count)
+ while idx < count:
+ bfs_queue.append(CastIOKitClass(ca.array[idx], 'IORegistryEntry *'))
+ idx += 1
+
+
+def FindMatchingServices(matching_name):
+ """
+ Finds registry entries that match the given string. Works similarly to:
+
+ io_iterator_t iter;
+ IOServiceGetMatchingServices(..., IOServiceMatching(matching_name), &iter);
+ while (( io_object_t next = IOIteratorNext(iter))) { ... }
+
+ Args:
+ matching_name (str): The class name to search for.
+
+ Yields:
+ core.cvalue.value: LLDB values that represent IORegistryEntry * for
+ each registry entry found.
+ """
+
+ # Check if the argument is valid
+ metaclasses = GetMetaClasses()
+ if matching_name not in metaclasses:
+ return
+ matching_metaclass = metaclasses[matching_name]
+
+ # An entry matches if it inherits from matching_metaclass
+ def matcher(entry):
+ # Get the class name of the entry and the associated metaclass
+ entry_name = GetRegistryEntryClassName(entry)
+ if entry_name in metaclasses:
+ entry_metaclass = metaclasses[entry_name]
+ return entry_metaclass.inheritsFrom(matching_metaclass)
+ else:
+ return False
+
+ # Search for entries
+ for entry in GetMatchingEntries(matcher):
+ yield entry
+
+
+def GetRegistryEntryParent(entry, iokit_plane=None):
+ """
+ Gets the parent entry of a registry entry.
+
+ Args:
+ entry (core.cvalue.value): A LLDB value representing a
+ IORegistryEntry *.
+ iokit_plane (core.cvalue.value, optional): A LLDB value representing a
+ IORegistryPlane *. By default, this method uses the IOService
+ plane.
+
+ Returns:
+ core.cvalue.value: A LLDB value representing a IORegistryEntry* that
+ is the parent entry of the entry argument in the specified plane.
+ Returns None if no entry could be found.
+ """
+ kParentSetIndex = 0
+ parent_key = None
+ if iokit_plane is None:
+ parent_key = kern.globals.gIOServicePlane.keys[kParentSetIndex]
+ else:
+ parent_key = plane.keys[kParentSetIndex]
+ parent_array = LookupKeyInOSDict(entry.fRegistryTable, parent_key)
+ parent_entry = None
+ if parent_array is not None:
+ idx = 0
+ ca = CastIOKitClass(parent_array, 'OSArray *')
+ count = unsigned(ca.count)
+ if count > 0:
+ parent_entry = CastIOKitClass(ca.array[0], 'IORegistryEntry *')
+ return parent_entry
+
+
+def GetRegistryEntryInterruptProperties(entry):
+ """
+ Get the interrupt properties of a registry entry.
+
+ Args:
+ entry (core.cvalue.value): A LLDB value representing a IORegistryEntry *.
+
+ Returns:
+ (bool, List[core.cvalue.value], List[str]): A tuple with the following
+ fields:
+ - First field (bool): Whether this entry has a non-null
+ IOPCIMSIMode.
+ - Second field (List[core.cvalue.value]): A list of LLDB values
+ representing OSData *. The OSData* pointer points to
+ interrupt vector data.
+ - Third field (List[str]): A list of strings representing the
+ interrupt controller names from the
+ IOInterruptControllers property.
+ """
+ INTERRUPT_SPECIFIERS_PROPERTY = "IOInterruptSpecifiers"
+ INTERRUPT_CONTROLLERS_PROPERTY = "IOInterruptControllers"
+ MSI_MODE_PROPERTY = "IOPCIMSIMode"
+
+ # Check IOInterruptSpecifiers
+ interrupt_specifiers = LookupKeyInPropTable(entry.fPropertyTable, INTERRUPT_SPECIFIERS_PROPERTY)
+ if interrupt_specifiers is not None:
+ interrupt_specifiers = CastIOKitClass(interrupt_specifiers, 'OSArray *')
+
+ # Check IOInterruptControllers
+ interrupt_controllers = LookupKeyInPropTable(entry.fPropertyTable, INTERRUPT_CONTROLLERS_PROPERTY)
+ if interrupt_controllers is not None:
+ interrupt_controllers = CastIOKitClass(interrupt_controllers, 'OSArray *')
+
+ # Check MSI mode
+ msi_mode = LookupKeyInPropTable(entry.fPropertyTable, MSI_MODE_PROPERTY)
+
+ result_vector_data = []
+ result_vector_cont = []
+ if interrupt_specifiers is not None and interrupt_controllers is not None:
+ interrupt_specifiers_array_count = unsigned(interrupt_specifiers.count)
+ interrupt_controllers_array_count = unsigned(interrupt_controllers.count)
+ # The array lengths should be the same
+ if interrupt_specifiers_array_count == interrupt_controllers_array_count and interrupt_specifiers_array_count > 0:
+ idx = 0
+ while idx < interrupt_specifiers_array_count:
+ # IOInterruptSpecifiers is an array of OSData *
+ vector_data = CastIOKitClass(interrupt_specifiers.array[idx], "OSData *")
+
+ # IOInterruptControllers is an array of OSString *
+ vector_cont = GetString(interrupt_controllers.array[idx])
+
+ result_vector_data.append(vector_data)
+ result_vector_cont.append(vector_cont)
+ idx += 1
+
+ return (msi_mode is not None, result_vector_data, result_vector_cont)
+
+
+class InterruptControllerDevice(object):
+ """Represents a IOInterruptController"""
+
+ def __init__(self, device, driver, base_vector_number, name):
+ """
+ Initialize a InterruptControllerDevice.
+
+ Args:
+ device (core.cvalue.value): The device object.
+ driver (core.cvalue.value): The driver object.
+ base_vector_number (int): The base interrupt vector.
+ name (str): The name of this interrupt controller.
+
+ Note:
+ Use the factory method makeInterruptControllerDevice to validate
+ properties.
+ """
+ self.device = device
+ self.driver = driver
+ self.name = name
+ self.base_vector_number = base_vector_number
+
+
+ def __str__(self):
+ """
+ String representation of this InterruptControllerDevice.
+ """
+ return " Name {}, base vector = {}, device = {}, driver = {}".format(
+ self.name, hex(self.base_vector_number), str(self.device), str(self.driver))
+
+ @staticmethod
+ def makeInterruptControllerDevice(device, driver):
+ """
+ Factory method to create a InterruptControllerDevice.
+
+ Args:
+ device (core.cvalue.value): The device object.
+ driver (core.cvalue.value): The driver object.
+
+ Returns:
+ InterruptControllerDevice: Returns an instance of
+ InterruptControllerDevice or None if the arguments do not have
+ the required properties.
+ """
+ BASE_VECTOR_PROPERTY = "Base Vector Number"
+ INTERRUPT_CONTROLLER_NAME_PROPERTY = "InterruptControllerName"
+ base_vector = LookupKeyInPropTable(device.fPropertyTable, BASE_VECTOR_PROPERTY)
+ if base_vector is None:
+ base_vector = LookupKeyInPropTable(driver.fPropertyTable, BASE_VECTOR_PROPERTY)
+ device_name = LookupKeyInPropTable(device.fPropertyTable, INTERRUPT_CONTROLLER_NAME_PROPERTY)
+ if device_name is None:
+ device_name = LookupKeyInPropTable(driver.fPropertyTable, INTERRUPT_CONTROLLER_NAME_PROPERTY)
+
+ if device_name is not None:
+ # Some interrupt controllers do not have a base vector number. Assume it is 0.
+ base_vector_number = 0
+ if base_vector is not None:
+ base_vector_number = unsigned(GetNumber(base_vector))
+ device_name = GetString(device_name)
+ # Construct object and return
+ return InterruptControllerDevice(device, driver, base_vector_number, device_name)
+ else:
+ # error case
+ return None
+
+
+def SearchInterruptControllerDrivers():
+ """
+ Search the IOKit registry for entries that match IOInterruptController.
+
+ Yields:
+ core.cvalue.value: A LLDB value representing a IORegistryEntry * that
+ inherits from IOInterruptController.
+ """
+ for entry in FindMatchingServices("IOInterruptController"):
+ # Get parent
+ parent = GetRegistryEntryParent(entry)
+
+ # Make the interrupt controller object
+ ic = InterruptControllerDevice.makeInterruptControllerDevice(parent, entry)
+
+ # Yield object
+ if ic is not None:
+ yield ic
+
+
+def LookupKeyInOSDict(osdict, key, comparer = None):
""" Returns the value corresponding to a given key in a OSDictionary
Returns None if the key was not found
"""
count = unsigned(osdict.count)
result = None
idx = 0
+
+ if not comparer:
+ # When comparer is specified, "key" argument can be of any type as "comparer" knows how to compare "key" to a key from "osdict".
+ # When comparer is not specified, key is of cpp_obj type.
+ key = getOSPtr(key)
while idx < count and result is None:
- if key == osdict.dictionary[idx].key:
+ if comparer is not None:
+ if comparer(key, osdict.dictionary[idx].key) == 0:
+ result = osdict.dictionary[idx].value
+ elif key == osdict.dictionary[idx].key:
result = osdict.dictionary[idx].value
idx += 1
return result
def GetString(string):
""" Returns the python string representation of a given OSString
"""
- out_string = "\"{0:s}\"".format(CastIOKitClass(string, 'OSString *').string)
+ out_string = "{0:s}".format(CastIOKitClass(string, 'OSString *').string)
return out_string
def GetNumber(num):
def GetDictionary(d):
""" Returns a string containing info about a given OSDictionary
"""
- out_string = "{"
+ if d is None:
+ return ""
+ out_string = "{\n"
idx = 0
count = unsigned(d.count)
while idx < count:
- obj = d.dictionary[idx].key
- out_string += GetObjectSummary(obj) + "="
- obj = d.dictionary[idx].value
+ key = d.dictionary[idx].key
+ value = d.dictionary[idx].value
+ out_string += " \"{}\" = {}\n".format(GetString(key), GetObjectSummary(value))
idx += 1
- out_string += GetObjectSummary(obj)
- if idx < count:
- out_string += ","
out_string += "}"
return out_string
def GetSet(se):
""" Returns a string containing info about a given OSSet
"""
- out_string += "[" + GetArray(se.members) + "]"
+ out_string = "[" + GetArray(se.members) + "]"
return out_string
def ReadIOPortInt(addr, numbytes, lcpu):
return True
+def GetRegistryPlane(plane_name):
+ """
+ Given plane_name, returns IORegistryPlane * object or None if there's no such registry plane
+ """
+ return LookupKeyInOSDict(kern.globals.gIORegistryPlanes, plane_name, CompareStringToOSSymbol)
+
+def DecodePreoslogSource(source):
+ """
+ Given preoslog source, return a matching string representation
+ """
+ source_to_str = {0 : "iboot"}
+ if source in source_to_str:
+ return source_to_str[source]
+ return "UNKNOWN"
+
+def GetPreoslogHeader():
+ """
+ Scan IODeviceTree for preoslog and return a python representation of it
+ """
+ edt_plane = GetRegistryPlane("IODeviceTree")
+ if edt_plane is None:
+ print "Couldn't obtain a pointer to IODeviceTree"
+ return None
+
+ # Registry API functions operate on "plane" global variable
+ global plane
+ prev_plane = plane
+ plane = edt_plane
+ chosen = FindRegistryObjectRecurse(kern.globals.gRegistryRoot, "chosen")
+ if chosen is None:
+ print "Couldn't obtain /chosen IORegistryEntry"
+ return None
+
+ memory_map = FindRegistryObjectRecurse(chosen, "memory-map")
+ if memory_map is None:
+ print "Couldn't obtain memory-map from /chosen"
+ return None
+
+ plane = prev_plane
+
+ mm_preoslog = LookupKeyInOSDict(memory_map.fPropertyTable, "preoslog", CompareStringToOSSymbol)
+ if mm_preoslog is None:
+ print "Couldn't find preoslog entry in memory-map"
+ return None
+
+ if mm_preoslog.length != 16:
+ print "preoslog entry in memory-map is malformed, expected len is 16, given len is {}".format(preoslog.length)
+ return None
+
+ data = cast(mm_preoslog.data, "dtptr_t *")
+ preoslog_paddr = unsigned(data[0])
+ preoslog_vaddr = kern.PhysToKernelVirt(preoslog_paddr)
+ preoslog_size = unsigned(data[1])
+
+ preoslog_header = PreoslogHeader()
+
+ # This structure defnition doesn't exist in xnu
+ """
+ typedef struct __attribute__((packed)) {
+ char magic[4];
+ uint32_t size;
+ uint32_t offset;
+ uint8_t source;
+ uint8_t wrapped;
+ char data[];
+ } preoslog_header_t;
+ """
+ preoslog_header_ptr = kern.GetValueFromAddress(preoslog_vaddr, "uint8_t *")
+ preoslog_header.magic = preoslog_header_ptr[0:4]
+ preoslog_header.source = DecodePreoslogSource(unsigned(preoslog_header_ptr[12]))
+ preoslog_header.wrapped = unsigned(preoslog_header_ptr[13])
+ preoslog_header_ptr = kern.GetValueFromAddress(preoslog_vaddr, "uint32_t *")
+ preoslog_header.size = unsigned(preoslog_header_ptr[1])
+ preoslog_header.offset = unsigned(preoslog_header_ptr[2])
+
+ for i in xrange(len(preoslog_header.valid_magic)):
+ c = chr(unsigned(preoslog_header.magic[i]))
+ if c != preoslog_header.valid_magic[i]:
+ string = "Error: magic doesn't match, expected {:.4s}, given {:.4s}"
+ print string.format(preoslog_header.valid_magic, preoslog_header.magic)
+ return None
+
+ if preoslog_header.size != preoslog_size:
+ string = "Error: size mismatch preoslog_header.size ({}) != preoslog_size ({})"
+ print string.format(preoslog_header.size, preoslog_size)
+ return None
+
+ preoslog_data_ptr = kern.GetValueFromAddress(preoslog_vaddr + 14, "char *")
+ preoslog_header.data = preoslog_data_ptr.sbvalue.GetPointeeData(0, preoslog_size)
+ return preoslog_header
+
+@lldb_command("showpreoslog")
+def showpreoslog(cmd_args=None):
+ """ Display preoslog buffer """
+
+ preoslog = GetPreoslogHeader()
+ if preoslog is None:
+ print "Error: couldn't obtain preoslog header"
+ return False
+
+ header = "".join([
+ "----preoslog log header-----\n",
+ "size - {} bytes\n",
+ "write offset - {:#x}\n",
+ "wrapped - {}\n",
+ "source - {}\n",
+ "----preoslog log start------"
+ ])
+
+ print header.format(preoslog.size, preoslog.offset, preoslog.wrapped, preoslog.source)
+
+ err = lldb.SBError()
+ if preoslog.wrapped > 0:
+ print preoslog.data.GetString(err, preoslog.offset + 1)
+ print preoslog.data.GetString(err, 0)
+ print "-----preoslog log end-------"
+ return True
projects / apple / xnu.git / blobdiff