vtrace.tools.win32heap

1 ''' 2 Windows heap allocation helper module 3 ''' 4 # Heap Flags 5 HEAP_NO_SERIALIZE = 0x00000001 6 HEAP_GROWABLE = 0x00000002 7 HEAP_GENERATE_EXCEPTIONS = 0x00000004 8 HEAP_ZERO_MEMORY = 0x00000008 9 HEAP_REALLOC_IN_PLACE_ONLY = 0x00000010 10 HEAP_TAIL_CHECKING_ENABLED = 0x00000020 11 HEAP_FREE_CHECKING_ENABLED = 0x00000040 12 HEAP_DISABLE_COALESCE_ON_FREE = 0x00000080 13 HEAP_CREATE_ALIGN_16 = 0x00010000 14 HEAP_CREATE_ENABLE_TRACING = 0x00020000 15 HEAP_CREATE_ENABLE_EXECUTE = 0x00040000 16 17 heap_flag_names = { 18 HEAP_NO_SERIALIZE:"HEAP_NO_SERIALIZE", 19 HEAP_GROWABLE:"HEAP_GROWABLE", 20 HEAP_GENERATE_EXCEPTIONS:"HEAP_GENERATE_EXCEPTIONS", 21 HEAP_ZERO_MEMORY:"HEAP_ZERO_MEMORY", 22 HEAP_REALLOC_IN_PLACE_ONLY:"HEAP_REALLOC_IN_PLACE_ONLY", 23 HEAP_TAIL_CHECKING_ENABLED:"HEAP_TAIL_CHECKING_ENABLED", 24 HEAP_FREE_CHECKING_ENABLED:"HEAP_FREE_CHECKING_ENABLED", 25 HEAP_DISABLE_COALESCE_ON_FREE:"HEAP_DISABLE_COALESCE_ON_FREE", 26 HEAP_CREATE_ALIGN_16:"HEAP_CREATE_ALIGN_16", 27 HEAP_CREATE_ENABLE_TRACING:"HEAP_CREATE_ENABLE_TRACING", 28 HEAP_CREATE_ENABLE_EXECUTE:"HEAP_CREATE_ENABLE_EXECUTE", 29 } 30 31 # Heap Chunk Flags 32 HEAP_ENTRY_BUSY = 0x01 33 HEAP_ENTRY_EXTRA_PRESENT = 0x02 34 HEAP_ENTRY_FILL_PATTERN = 0x04 35 HEAP_ENTRY_VIRTUAL_ALLOC = 0x08 36 HEAP_ENTRY_LAST_ENTRY = 0x10 37 HEAP_ENTRY_SETTABLE_FLAG1 = 0x20 38 HEAP_ENTRY_SETTABLE_FLAG2 = 0x40 39 HEAP_ENTRY_SETTABLE_FLAG3 = 0x80 40 41 42

43 -def reprHeapFlags(flags):

44 ret = [] 45 if flags & HEAP_ENTRY_BUSY: 46 ret.append("BUSY") 47 if flags & HEAP_ENTRY_FILL_PATTERN: 48 ret.append("FILL") 49 if flags & HEAP_ENTRY_LAST_ENTRY: 50 ret.append("LAST") 51 if len(ret): 52 return "|".join(ret) 53 return "NONE"

54

55 -class HeapCorruptionException(Exception):

56 - def __init__(self, heap, segment, prevchunk, badchunk):

57 prevaddr = 0 58 badaddr = 0 59 if prevchunk != None: 60 prevaddr = prevchunk.address 61 if badchunk != None: 62 badaddr = badchunk.address 63 Exception.__init__(self, "Prev Chunk: 0x%.8x Bad Chunk: 0x%.8x" % (prevaddr, badaddr)) 64 self.heap = heap 65 self.segment = segment 66 self.prevchunk = prevchunk 67 self.badchunk = badchunk

68

69 -class FreeListCorruption(Exception):

70 - def __init__(self, heap, index, prevchunk, badchunk):

71 prevaddr = 0 72 badaddr = 0 73 if prevchunk != None: 74 prevaddr = prevchunk.address 75 if badchunk != None: 76 badaddr = badchunk.address 77 Exception.__init__(self, "Index: %d Prev Chunk: 0x%.8x Bad Chunk: 0x%.8x" % (index, prevaddr, badaddr)) 78 self.heap = heap 79 self.index = index 80 self.prevchunk = prevchunk 81 self.badchunk = badchunk

82

83 -class ChunkNotFound(Exception):

84 pass

85

86 -def getHeapSegChunk(trace, address):

87 """ 88 Find and return the heap, segment, and chunk for the given addres 89 (or exception). 90 """ 91 for heap in getHeaps(trace): 92 93 for seg in heap.getSegments(): 94 95 segstart = seg.address 96 segend = seg.getSegmentEnd() 97 98 if address < segstart or address > segend: 99 continue 100 101 for chunk in seg.getChunks(): 102 a = chunk.address 103 b = chunk.address + len(chunk) 104 if address >= a and address < b: 105 return heap,seg,chunk 106 107 raise ChunkNotFound("No Chunk Found for 0x%.8x" % address)

108

109 -def getHeaps(trace):

110 """ 111 Get the win32 heaps (returns a list of Win32Heap objects) 112 """ 113 ret = [] 114 pebaddr = trace.getMeta("PEB") 115 peb = trace.getStruct("ntdll.PEB", pebaddr) 116 heapcount = int(peb.NumberOfHeaps) 117 hlist = trace.readMemoryFormat(long(peb.ProcessHeaps), "<"+('P'*heapcount)) 118 for haddr in hlist: 119 ret.append(Win32Heap(trace, haddr)) 120 return ret

121

122 -class Win32Heap:

123

124 - def __init__(self, trace, address):

125 self.address = address 126 self.trace = trace 127 self.heap = trace.getStruct("ntdll.HEAP", address) 128 self._win7_heap = False 129 self.heapenc = None 130 if self.heap.vsHasField('Encoding'): 131 self.heapenc = self.heap.Encoding 132 self._win7_heap = True 133 self.seglist = None 134 self.ucrdict = None

135

136 - def hasLookAside(self):

137 """ 138 Does this heap have a lookaside? 139 """ 140 if not self.heap.Flags & HEAP_GROWABLE: 141 return False 142 if self.heap.Flags & HEAP_NO_SERIALIZE: 143 return False 144 return True

145

146 - def getUCRDict(self):

147 ''' 148 Retrieve a dictionary of <ucr_address>:<ucr_size> items. 149 150 (If this windows version doesn't support UCRs, the dict will be empty) 151 ''' 152 if self.ucrdict == None: 153 self.ucrdict = {} 154 if self.heap.vsHasField('UCRList'): 155 listhead_va = self.address + self.heap.vsGetOffset('UCRList') 156 for lva in self._getListEntries(listhead_va): 157 ucrent = self.trace.getStruct('ntdll.HEAP_UCR_DESCRIPTOR', lva) 158 if ucrent.Size != 0: 159 self.ucrdict[ucrent.Address] = ucrent.Size 160 161 return self.ucrdict

162

163 - def _win7ParseSegments(self):

164 # Address doesn't matter for the below call 165 heapseg = self.trace.getStruct('ntdll.HEAP_SEGMENT', self.address) 166 167 listhead = self.address + self.heap.vsGetOffset('SegmentList') 168 # Negative offset from segment list entry to segment 169 segoff = heapseg.vsGetOffset('SegmentListEntry') 170 entry = self.heap.SegmentList.Flink 171 while entry != listhead: 172 seg = Win32Segment(self.trace, self, entry - segoff) 173 self.seglist.append(seg) 174 entry = self.trace.readMemoryFormat(entry, '<P')[0]

175

176 - def getSegments(self):

177 """ 178 Return a list of Win32Segment objects. 179 """ 180 if self.seglist == None: 181 self.seglist = [] 182 183 # Windows 7 style heap segment list 184 if self.heap.vsHasField('SegmentList'): 185 self._win7ParseSegments() 186 187 else: 188 for i in range(long(self.heap.LastSegmentIndex)+1): 189 sa = self.heap.Segments[i] 190 self.seglist.append(Win32Segment(self.trace, self, long(sa))) 191 192 return self.seglist

193

194 - def getLookAsideLists(self):

195 """ 196 Return a list of the lookaside list for this heap 197 """ 198 if not self.hasLookAside(): 199 raise Exception("Heap at 0x%.8x has no lookaside!" % (self.address)) 200 # Look aside lists are 128 pointer slots in a chunk pointed 201 # to by the FrontEndHeap field in the heap structure. 202 #fetype = self.heap.FrontEndHeapType 203 204 laside = self.heap.FrontEndHeap 205 ret = [] 206 for i in range(128): 207 slot = laside + (i * 0x30) 208 bucket = [] 209 base = self.trace.readMemoryFormat(slot, "<I")[0] 210 while base != 0: 211 chunk = Win32Chunk(self.trace, self, base-8) 212 bucket.append(chunk) 213 base,blink = chunk.getFlinkBlink() 214 ret.append(bucket) 215 return ret

216

217 - def _getListEntries(self, addr, listhead=True):

218 ret = [] 219 if not listhead: 220 ret.append(addr) 221 le = self.trace.getStruct('ntdll.LIST_ENTRY', addr) 222 while le.Flink != addr: 223 ret.append(le.Flink) 224 le = self.trace.getStruct('ntdll.LIST_ENTRY', le.Flink) 225 return ret

226

227 - def _win7FreeLists(self):

228 #print 'Windows 7 Free List Parsing Fixme!' 229 return []

230

231 - def getFreeLists(self):

232 """ 233 Return a list of the free lists in this heap. 234 (Not including look-aside) 235 """ 236 ret = [] 237 foff = self.heap.vsGetOffset("FreeLists") 238 if self._win7_heap: 239 return self._win7FreeLists() 240 241 for i in range(128): 242 le = self.heap.FreeLists[i] 243 bucket = [] 244 base = self.address + foff + (i*8) 245 246 addr = le.Flink 247 248 while addr != base: 249 250 # If we die here, the guy before us was dorked. 251 try: 252 chunk = Win32Chunk(self.trace, self, addr-8) 253 except Exception, e: 254 chunk = bucket[-1] 255 pchunk = None 256 if len(bucket) >= 2: 257 pchunk = bucket[-2] 258 raise FreeListCorruption(self, i, pchunk, chunk) 259 260 # Check our back pointer 261 flink,blink = chunk.getFlinkBlink() 262 if len(bucket): 263 pchunk = bucket[-1] 264 if blink != pchunk.getDataAddress(): 265 raise FreeListCorruption(self, i, pchunk, chunk) 266 267 addr = flink 268 bucket.append(chunk) 269 270 ret.append(bucket) 271 return ret

272

273 - def getFlagNames(self):

274 ret = [] 275 for k,v in heap_flag_names.items(): 276 if self.heap.Flags & k: 277 ret.append(v) 278 return ret

279

280 - def __repr__(self):

281 fnames = "|".join(self.getFlagNames()) 282 return "heap: 0x%.8x flags:%s" % (self.address, fnames)

283

284 -class Win32Segment:

285 - def __init__(self, trace, heap, address):

286 self.trace = trace 287 self.heap = heap 288 self.address = address 289 self.seg = trace.getStruct("ntdll.HEAP_SEGMENT", address) 290 #FIXME segments can specify chunk Size granularity 291 self.chunks = None 292 self.segend = self.address + (self.seg.NumberOfPages * 4096)

293

294 - def getSegmentEnd(self):

295 return self.segend

296

297 - def getChunks(self):

298 if self.chunks == None: 299 self.chunks = [] 300 addr = self.address 301 lastchunk = None 302 ucrdict = self.heap.getUCRDict() 303 304 while True: 305 306 # Skip any uncommited ranges... 307 usize = ucrdict.get(addr) 308 if usize != None: 309 lastchunk = None 310 addr += usize 311 continue 312 313 # Since an un-commited range may put us past the 314 # lastblock (segend) we must double check... 315 if addr >= self.segend: 316 break 317 318 chunk = Win32Chunk(self.trace, self.heap, addr) 319 320 self.chunks.append(chunk) 321 #if lastchunk != None: 322 #if lastchunk.chunk.Size != chunk.chunk.PreviousSize: 323 #print 'last size:',lastchunk.chunk.Size,'prev',chunk.chunk.PreviousSize 324 #raise HeapCorruptionException(self.heap, self, lastchunk, chunk) 325 326 if chunk.isLast(): 327 break 328 329 lastchunk = chunk 330 addr += len(chunk) 331 return self.chunks

332

333 - def getLastChunk(self):

334 va = self.seg.LastEntryInSegment 335 return Win32Chunk(self.trace, self.heap, va)

336

337 -class Win32Chunk:

338 - def __init__(self, trace, heap, address):

339 self.trace = trace 340 self.heap = heap 341 self.address = address 342 self.chunk = trace.getStruct("ntdll.HEAP_ENTRY", address) 343 344 # Decode the heap chunk if needed... 345 if self.heap.heapenc: 346 self.chunk ^= self.heap.heapenc

347

348 - def __repr__(self):

349 return "HeapChunk: 0x%.8x (%d) %s" % (self.address, len(self),self.reprFlags())

350

351 - def __len__(self):

352 return int(self.chunk.Size) * 8

353

354 - def isLast(self):

355 return bool(int(self.chunk.Flags) & HEAP_ENTRY_LAST_ENTRY)

356

357 - def isBusy(self):

358 return bool(int(self.chunk.Flags) & HEAP_ENTRY_BUSY)

359

360 - def getDataAddress(self):

361 return self.address + len(self.chunk)

362

363 - def getDataSize(self):

364 return len(self) - len(self.chunk)

365

366 - def getDataBytes(self, maxsize=None):

367 size = self.getDataSize() 368 if maxsize != None: 369 size = min(size, maxsize) 370 return self.trace.readMemory(self.getDataAddress(), size)

371

372 - def getFlinkBlink(self):

373 return self.trace.readMemoryFormat(self.getDataAddress(), "<PP")

374

375 - def reprFlags(self):

376 return reprHeapFlags(int(self.chunk.Flags))

377

Source Code for Module vtrace.tools.win32heap