Source Code for Module rekall.plugins.windows.interactive.structs

  1  # Rekall Memory Forensics 
  2  # Copyright 2014 Google Inc. All Rights Reserved. 
  3  # 
  4  # This program is free software; you can redistribute it and/or modify 
  5  # it under the terms of the GNU General Public License as published by 
  6  # the Free Software Foundation; either version 2 of the License, or (at 
  7  # your option) any later version. 
  8  # 
  9  # This program is distributed in the hope that it will be useful, but 
 10  # WITHOUT ANY WARRANTY; without even the implied warranty of 
 11  # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 
 12  # General Public License for more details. 
 13  # 
 14  # You should have received a copy of the GNU General Public License 
 15  # along with this program; if not, write to the Free Software 
 16  # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 
 17  # 
 18   
 19  """Interactive plugins. 
 20   
 21  This file contains a bunch of plugins which are useful when interactively 
 22  examining a memory image. 
 23  """ 
 24  import itertools 
 25   
 26  # pylint: disable=protected-access 
 27  from rekall import obj 
 28  from rekall.plugins.windows import common 
 29  from rekall_lib import utils 
 30   
 31   
 32 -class AnalyzeStruct(common.WindowsCommandPlugin): 
 33      """A plugin to analyze a memory location.""" 
 34      name = "analyze_struct" 
 35   
 36      __args = [ 
 37          dict(name="offset", positional=True, type="SymbolAddress", 
 38               required=True, 
 39               help="A virtual address to analyze."), 
 40   
 41          dict(name="search", type="IntParser", default=0x100, 
 42               help="How far back to search for pool tag."), 
 43   
 44          dict(name="size", type="IntParser", default=0x100, 
 45               help="How many elements to identify."), 
 46      ] 
 47   
 48      table_header = [ 
 49          dict(name="divider", type="Divider"), 
 50          dict(name="offset", style="address"), 
 51          dict(name="pool_offset", style="address"), 
 52          dict(name="content") 
 53      ] 
 54   
 55 -    def SearchForPoolHeader(self, offset, search=0x100): 
 56          """Search backwards from offset for a pool header.""" 
 57          pool_alignment = self.session.profile.get_constant("PoolAlignment") 
 58          offset = int(offset) - offset % pool_alignment 
 59   
 60          # Cant use xrange() on windows since it must fit into a long. 
 61          for o in itertools.count(offset, -pool_alignment): 
 62              if o < offset-search: 
 63                  break 
 64   
 65              pool_header = self.session.profile._POOL_HEADER(o) 
 66   
 67              # If this is the pool header for this allocation it must be big 
 68              # enough to contain it. 
 69              if pool_header.BlockSize < (offset - o) / pool_alignment + 1: 
 70                  continue 
 71   
 72              #if not pool_header.PoolType.is_valid(): 
 73              #    continue 
 74   
 75              # Verify it. 
 76              if pool_header.PreviousSize > 0: 
 77                  previous_pool_header = self.session.profile._POOL_HEADER( 
 78                      o - pool_alignment * pool_header.PreviousSize) 
 79   
 80                  if previous_pool_header.BlockSize == pool_header.PreviousSize: 
 81                      return pool_header 
 82   
 83              # Check the next allocation. 
 84              next_pool_header = self.session.profile._POOL_HEADER( 
 85                  o + pool_alignment * pool_header.BlockSize) 
 86   
 87              if next_pool_header.PreviousSize == pool_header.BlockSize: 
 88                  return pool_header 
 89   
 90          return obj.NoneObject("No pool tag found") 
 91   
 92 -    def GuessMembers(self, offset, size=0x100, search=0x100): 
 93          offset = int(offset) 
 94          resolver = self.session.address_resolver 
 95          result = [] 
 96   
 97          for member in self.session.profile.Array(offset, target="Pointer", 
 98                                                   count=size/8): 
 99              address_info = ["Data:%#x" % member.v()] 
100              relative_offset = member.obj_offset - offset 
101              result.append((relative_offset, address_info)) 
102   
103              # Try to find pointers to known pool allocations. 
104              pool = self.SearchForPoolHeader(member.v(), search=search) 
105              if pool: 
106                  address_info.append("Tag:%s" % pool.Tag) 
107                  proc = pool.m("ProcessBilled") 
108                  # Does the tag refer to a real _EPROCESS? If so it must have a 
109                  # valid environment block (and a corresponding address space). 
110                  if proc.Peb: 
111                      address_info.append("ProcessBilled:%s" % proc.name) 
112   
113                  address_info.append("@ %#x (%#x)" % (member.v(), pool.size)) 
114   
115              else: 
116                  # Look for pointers to global symbols. 
117                  sym_offset, symbol = resolver.get_nearest_constant_by_address( 
118                      member.v()) 
119   
120                  if symbol and sym_offset == member.v(): 
121                      address_info.append("Const:%s" % ", ".join(symbol)) 
122   
123              # Check for _LIST_ENTRYs 
124              list_member = member.cast("_LIST_ENTRY") 
125              if list_member.obj_offset == list_member.Flink.Blink.v(): 
126                  address_info.append("_LIST_ENTRY") 
127                  address_info.append("@ %#x" % list_member.Flink.v()) 
128   
129              if list_member.obj_offset == list_member.Flink.v(): 
130                  address_info.append("Empty") 
131   
132          return result 
133   
134 -    def collect(self): 
135          pool_offset = None 
136          pool_header = self.SearchForPoolHeader( 
137              self.plugin_args.offset, search=self.plugin_args.search) 
138   
139          if pool_header: 
140              name = (pool_header.m("ProcessBilled").name or 
141                      str(pool_header.Tag).encode("string-escape")) 
142   
143              yield dict(divider=("{0:#x} is inside pool allocation with " 
144                                  "tag '{1}' ({2:#x}) and size {3:#x}".format( 
145                                      self.plugin_args.offset, 
146                                      name, pool_header, pool_header.size))) 
147   
148          for relative_offset, info in self.GuessMembers( 
149                  self.plugin_args.offset, size=self.plugin_args.size, 
150                  search=self.plugin_args.search): 
151   
152              if pool_header: 
153                  pool_offset = (self.plugin_args.offset + relative_offset - 
154                                 pool_header.obj_offset) 
155   
156              yield dict(offset=relative_offset, 
157                         pool_offset=pool_offset, 
158                         content=" ".join( 
159                  [utils.SmartStr(x).encode("string-escape") for x in info])) 
160