1""" 2Wrappers around globals and caches to service the kmem package 3""" 4from abc import ABCMeta, abstractmethod 5from collections import namedtuple 6from core import ( 7 caching, 8 gettype, 9 lldbwrap, 10) 11from ctypes import c_int64 12 13class MemoryRange(namedtuple('MemoryRange', ['start', 'end'])): 14 @property 15 def size(self): 16 start, end = self 17 return end - start 18 19 def contains(self, addr): 20 start, end = self 21 return start <= addr < end 22 23 def __repr__(self): 24 return "{0.__class__.__name__}[{0.start:#x}, {0.end:#x})".format(self) 25 26 27class VMPointerUnpacker(object): 28 """ 29 Pointer unpacker for pointers packed with VM_PACK_POINTER() 30 """ 31 def __init__(self, target, param_var): 32 params = target.chkFindFirstGlobalVariable(param_var) 33 self.base_relative = params.xGetScalarByName('vmpp_base_relative') 34 self.bits = params.xGetScalarByName('vmpp_bits') 35 self.shift = params.xGetScalarByName('vmpp_shift') 36 self.base = params.xGetScalarByName('vmpp_base') 37 38 def unpack(self, packed): 39 """ 40 Unpacks an address according to the VM_PACK_POINTER() scheme 41 42 @param packed (int) 43 The packed value to unpack 44 45 @returns (int) 46 The unpacked address 47 """ 48 49 if not packed: 50 return None 51 52 if self.base_relative: 53 addr = (packed << self.shift) + self.base 54 else: 55 bits = self.bits 56 shift = self.shift 57 addr = c_int64(packed << (64 - bits)).value 58 addr >>= 64 - bits - shift 59 60 return addr & 0xffffffffffffffff 61 62 def unpack_value(self, sbv): 63 """ 64 Conveniency wrapper for self.unpack(sbv.chkGetValueAsUnsigned()) 65 """ 66 return self.unpack(sbv.chkGetValueAsUnsigned()) 67 68 69class KMem(object, metaclass=ABCMeta): 70 """ 71 Singleton class that holds various important information 72 that is needed to make sense of the kernel memory layout, 73 heap data structures, globals, ... 74 """ 75 76 _HEAP_NAMES = [ "", "shared.", "data.", "" ] 77 78 @staticmethod 79 def _parse_range(zone_info_v, name): 80 """ 81 Create a tuple representing a range (min_address, max_address, size) 82 """ 83 range_v = zone_info_v.chkGetChildMemberWithName(name) 84 left = range_v.xGetIntegerByName('min_address') 85 right = range_v.xGetIntegerByName('max_address') 86 return MemoryRange(left, right) 87 88 def __init__(self, target): 89 self.target = target 90 91 # 92 # Cache some globals everyone needs 93 # 94 self.page_shift = target.chkFindFirstGlobalVariable('page_shift').xGetValueAsInteger() 95 self.page_size = 1 << self.page_shift 96 self.page_mask = self.page_size - 1 97 98 phase_v = target.chkFindFirstGlobalVariable('startup_phase') 99 self.phase = phase_v.xGetValueAsInteger() 100 self.phases = set( 101 e.GetName()[len('STARTUP_SUB_'):] 102 for e in phase_v.GetType().get_enum_members_array() 103 if e.GetValueAsUnsigned() <= self.phase 104 ) 105 106 # 107 # Setup the number of CPUs we have 108 # 109 self.ncpus = target.chkFindFirstGlobalVariable('zpercpu_early_count').xGetValueAsInteger() 110 self.master_cpu = target.chkFindFirstGlobalVariable('master_cpu').xGetValueAsInteger() 111 self.zcpus = range(self.ncpus) if 'ZALLOC' in self.phases else (self.master_cpu, ) 112 self.pcpus = range(self.ncpus) if 'PERCPU' in self.phases else (self.master_cpu, ) 113 114 # 115 # Load all the ranges we will need 116 # 117 zone_info = target.chkFindFirstGlobalVariable('zone_info') 118 self.meta_range = self._parse_range(zone_info, 'zi_meta_range') 119 self.bits_range = self._parse_range(zone_info, 'zi_bits_range') 120 self.zone_range = self._parse_range(zone_info, 'zi_map_range') 121 try: 122 self.pgz_range = self._parse_range(zone_info, 'zi_pgz_range') 123 self.pgz_bt = target.chkFindFirstGlobalVariable('pgz_backtraces').xDereference() 124 except: 125 self.pgz_range = MemoryRange(0, 0) 126 self.pgz_bt = None 127 128 kmem_ranges = target.chkFindFirstGlobalVariable('kmem_ranges') 129 count = kmem_ranges.GetByteSize() // target.GetAddressByteSize() 130 addresses = target.xIterAsUInt64(kmem_ranges.GetLoadAddress(), count) 131 self.kmem_ranges = [ 132 MemoryRange(next(addresses), next(addresses)) 133 for i in range(0, count, 2) 134 ] 135 136 iokit_mach_vm_range = target.chkFindFirstGlobalVariable('gIOKitPageableFixedRange') 137 self.iokit_range = MemoryRange( 138 start=iokit_mach_vm_range.xGetIntegerByName('min_address'), 139 end=iokit_mach_vm_range.xGetIntegerByName('max_address'), 140 ) 141 142 # 143 # And other important globals 144 # 145 self.stext = target.chkFindFirstGlobalVariable('vm_kernel_stext').xGetValueAsInteger() 146 self.num_zones = target.chkFindFirstGlobalVariable('num_zones').xGetValueAsInteger() 147 self.mag_size = target.chkFindFirstGlobalVariable('_zc_mag_size').xGetValueAsInteger() 148 self.zone_array = target.chkFindFirstGlobalVariable('zone_array') 149 self.zsec_array = target.chkFindFirstGlobalVariable('zone_security_array') 150 151 self.kernel_map = target.chkFindFirstGlobalVariable('kernel_map').Dereference() 152 self.vm_kobject = target.chkFindFirstGlobalVariable('kernel_object_store') 153 154 # 155 # Cache some crucial types used for memory walks 156 # 157 self.zpm_type = gettype('struct zone_page_metadata') 158 self.vm_map_type = gettype('struct _vm_map') 159 self.vmo_type = self.vm_kobject.GetType() 160 161 # 162 # Recognize whether the target is any form of KASAN kernel. 163 # 164 if target.FindFirstGlobalVariable('kasan_enabled').IsValid(): 165 self.kasan = True 166 self.kasan_tbi = target.FindFirstGlobalVariable('kasan_tbi_enabled').IsValid() 167 self.kasan_classic = not self.kasan_tbi 168 else: 169 self.kasan = False 170 self.kasan_tbi = False 171 self.kasan_classic = False 172 173 # 174 # VM_PACK_POINTER Unpackers 175 # 176 self.kn_kq_packing = VMPointerUnpacker(target, 'kn_kq_packing_params') 177 self.vm_page_packing = VMPointerUnpacker(target, 'vm_page_packing_params') 178 try: 179 self.rwlde_caller_packing = VMPointerUnpacker(target, 'rwlde_caller_packing_params') 180 except ValueError: 181 # 182 # Release kernel doesn't define DEBUG_RW thus rwlde_caller_packing_params is compiled out 183 # 184 self.rwlde_caller_packing = None 185 186 self.c_slot_packing = VMPointerUnpacker(target, 'c_slot_packing_params') 187 188 @staticmethod 189 @caching.cache_statically 190 def get_shared(target=None): 191 """ 192 Returns a shared instance of the class 193 """ 194 195 arch = target.triple[:target.triple.find('-')] 196 197 if arch.startswith('arm64e'): 198 return _KMemARM64e(target) 199 elif arch.startswith('arm64'): 200 return _KMemARM64(target) 201 elif arch.startswith('x86_64'): 202 return _KMemX86(target) 203 else: 204 raise RuntimeError("Unsupported architecture: {}".format(arch)) 205 206 def iter_addresses(self, iterable): 207 """ 208 Conveniency wrapper to transform a list of integer to addresses 209 """ 210 return (self.make_address(a) for a in iterable) 211 212 # 213 # Abstract per-arch methods 214 # 215 216 @property 217 @abstractmethod 218 def has_ptrauth(self): 219 """ whether this target has ptrauth """ 220 221 pass 222 223 @abstractmethod 224 def PERCPU_BASE(self, cpu): 225 """ 226 Returns the per-cpu base for a given CPU number 227 228 @param cpu (int) 229 A CPU number 230 231 @returns (int) 232 The percpu base for this CPU 233 """ 234 235 pass 236 237 @abstractmethod 238 def make_address(self, addr): 239 """ 240 Make an address out of an integer 241 242 @param addr (int) 243 An address to convert 244 245 @returns (int) 246 """ 247 248 pass 249 250 251class _KMemARM64(KMem): 252 """ 253 Specialization of KMem for arm64 254 """ 255 256 def __init__(self, target): 257 super().__init__(target) 258 259 self.arm64_CpuDataEntries = target.chkFindFirstGlobalVariable('CpuDataEntries') 260 self.arm64_BootCpuData = target.chkFindFirstGlobalVariable('percpu_slot_cpu_data') 261 self.arm64_t1sz = target.chkFindFirstGlobalVariable('gT1Sz').xGetValueAsInteger() 262 self.arm64_sign_mask = 1 << (63 - self.arm64_t1sz) 263 264 @property 265 def has_ptrauth(self): 266 return False 267 268 def PERCPU_BASE(self, cpu): 269 cpu_data = self.arm64_CpuDataEntries.chkGetChildAtIndex(cpu) 270 boot_vaddr = self.arm64_BootCpuData.GetLoadAddress() 271 272 return cpu_data.xGetIntegerByName('cpu_data_vaddr') - boot_vaddr 273 274 def make_address(self, addr): 275 sign_mask = self.arm64_sign_mask 276 addr = addr & (sign_mask + sign_mask - 1) 277 return ((addr ^ sign_mask) - sign_mask) & 0xffffffffffffffff 278 279 280class _KMemARM64e(_KMemARM64): 281 """ 282 Specialization of KMem for arm64e 283 """ 284 285 @property 286 def has_ptrauth(self): 287 return True 288 289 290class _KMemX86(KMem): 291 """ 292 Specialization of KMem for Intel 293 """ 294 295 def __init__(self, target): 296 super().__init__(target) 297 298 self.intel_cpu_data = target.chkFindFirstGlobalVariable('cpu_data_ptr') 299 300 @property 301 def has_ptrauth(self): 302 return False 303 304 def PERCPU_BASE(self, cpu): 305 cpu_data = self.intel_cpu_data.chkGetChildAtIndex(cpu) 306 return cpu_data.xGetIntegerByName('cpu_pcpu_base') 307 308 def make_address(self, addr): 309 return addr 310 311 312class PERCPUValue(object): 313 """ 314 Provides an enumerator for a percpu value 315 """ 316 317 def __init__(self, name, target = None): 318 """ 319 @param name (str) 320 The percpu slot name 321 322 @param target (SBTarget or None) 323 """ 324 325 self.kmem = KMem.get_shared() 326 self.sbv = self.kmem.target.chkFindFirstGlobalVariable('percpu_slot_' + name) 327 328 def __getitem__(self, cpu): 329 if cpu in self.kmem.pcpus: 330 sbv = self.sbv 331 addr = sbv.GetLoadAddress() + self.kmem.PERCPU_BASE(cpu) 332 return sbv.chkCreateValueFromAddress(sbv.GetName(), addr, sbv.GetType()) 333 raise IndexError 334 335 def __iter__(self): 336 return (item[1] for items in self.items()) 337 338 def items(self): 339 """ 340 Iterator of (cpu, SBValue) tuples for the given PERCPUValue 341 """ 342 343 kmem = self.kmem 344 sbv = self.sbv 345 name = sbv.GetName() 346 ty = sbv.GetType() 347 addr = sbv.GetLoadAddress() 348 349 return ( 350 (cpu, sbv.chkCreateValueFromAddress(name, addr + kmem.PERCPU_BASE(cpu), ty)) 351 for cpu in kmem.pcpus 352 ) 353 354__all__ = [ 355 KMem.__name__, 356 MemoryRange.__name__, 357 PERCPUValue.__name__, 358] 359