Fixed some failing tests.

test_data/ubuntu_8.04/tests.config

@@ -10,4 +10,7 @@ commandline = dt task_struct
 commandline = dump 0xffff881f3d00
 
 [TestVtoP]
-commandline = vtop 0xffff881f3d00
+commandline = vtop 0xffff881f3d00
+
+[TestDisassemble]
+func = 0xffff802d4670

tools/testing/test_suite.py

@@ -82,8 +82,8 @@
 ====================
 
 Since the goal of this test suite is only to detect regressions, there is no
-need to write specialized test for each plugin. If not specific test for a
-plugin is found, the test suite will simple create one based on the
+need to write specialized test for each plugin. If no specific test for a
+plugin is found, the test suite will simply create one based on the
 testlib.SimpleTestCase() class - i.e. it just literally compares the output of
 the plugin. In most cases this is what we want.
 
@@ -97,7 +97,7 @@ class TestCheckTaskFops(testlib.SimpleTestCase):
         commandline="check_task_fops --all"
         )
 
-More complex example may use custom or more sophisticated methods for comparing
+More complex examples may use custom or more sophisticated methods for comparing
 the plugin output. For example the testlib.HashChecker() base class will ensure
 that all files produced by a plugin retain their hashes between executions:
 
@@ -148,6 +148,11 @@ def __init__(self, argv=None):
             logging.getLogger().setLevel(logging.DEBUG)
         self.renderer = renderer.TextRenderer()
 
+        # Some stats.
+        self.successes = 0
+        self.failures = 0
+        self.rebuilt = 0
+
     def ProcessCommandLineArgs(self, argv=None):
         parser = argparse.ArgumentParser()
 
@@ -243,6 +248,8 @@ def BuildBaseLineTask(self, config_options, plugin_cls):
                 self.renderer.color("REBUILT", foreground="YELLOW"),
                 baseline_data["time_used"])
 
+            self.rebuilt += 1
+
     def __enter__(self):
         self.temp_directory = tempfile.mkdtemp()
         return self
@@ -320,12 +327,16 @@ def RunTests(self):
                 continue
 
             # Retrieve the configured options if they exist.
-            config_options = dict(config.items("DEFAULT"))
-            config_options.update(plugin_cls.PARAMETERS)
+            config_options = plugin_cls.PARAMETERS.copy()
+
+            # Defaults section overrides the PARAMETERS attribute.
+            config_options.update(dict(config.items("DEFAULT")))
+
             config_options["test_class"] = plugin_cls.__name__
             if config.has_section(plugin_cls.__name__):
                 config_options.update(dict(config.items(plugin_cls.__name__)))
 
+
             # Try to get the previous baseline file.
             baseline_filename = os.path.join(
                 self.test_directory, plugin_cls.__name__)
@@ -381,6 +392,8 @@ def RunTestCase(self, config_options, plugin_cls, baseline_data):
                     self.renderer.color("PASS", foreground="GREEN"),
                     current_run.get("time_used", 0),
                     baseline_data.get("time_used", 0))
+                self.successes += 1
+
             else:
                 # Store the current run someplace for closer inspection.
                 with tempfile.NamedTemporaryFile(
@@ -393,16 +406,22 @@ def RunTestCase(self, config_options, plugin_cls, baseline_data):
                         current_run.get("time_used", 0),
                         baseline_data.get("time_used", 0),
                         fd.name)
+                    self.failures += 1
 
                 if self.FLAGS.verbose:
                     for test_case, error in result.errors + result.failures:
                         self.renderer.write("Error in %s: %s" % (
                                 plugin_cls.__name__, error))
 
 def main(argv):
+    start = time.time()
     with VolatilityTester() as tester:
         tester.RunTests()
 
+    tester.renderer.write(
+        "Completed %s tests (%s passed, %s failed, %s rebuild) in %s Seconds.\n" %
+        (tester.successes + tester.failures, tester.successes, tester.failures,
+         tester.rebuilt, int(time.time() - start)))
 
 if __name__ == "__main__":
     main(sys.argv)

volatility/addrspace.py

@@ -290,41 +290,34 @@ class PagedReader(BaseAddressSpace):
     PAGE_SIZE = 0x1000
     __abstract = True
 
-    def _read_chunk(self, vaddr, length, pad=False):
+    def _read_chunk(self, vaddr, length):
         """
         Read bytes from a virtual address.
 
         Args:
           vaddr: A virtual address to read from.
           length: The number of bytes to read.
-          pad: If set, pad unavailable data with nulls.
 
         Returns:
-          As many bytes as can be read within this page, or a NoneObject() if we
-          are not padding and the address is invalid.
+          As many bytes as can be read within this page.
         """
         to_read = min(length, self.PAGE_SIZE - (vaddr % self.PAGE_SIZE))
         paddr = self.vtop(vaddr)
         if paddr is None:
-            if pad:
-                return "\x00" * to_read
-            else:
-                return None
+            return "\x00" * to_read
 
         return self.base.read(paddr, to_read)
 
-    def _read_bytes(self, vaddr, length, pad):
+    def read(self, vaddr, length):
         """
         Read 'length' bytes from the virtual address 'vaddr'.
-        The 'pad' parameter controls whether unavailable bytes
-        are padded with zeros.
         """
         vaddr, length = int(vaddr), int(length)
 
         result = ''
 
         while length > 0:
-            buf = self._read_chunk(vaddr, length, pad=pad)
+            buf = self._read_chunk(vaddr, length)
             if not buf: break
 
             result += buf
@@ -333,13 +326,6 @@ def _read_bytes(self, vaddr, length, pad):
 
         return result
 
-    def read(self, vaddr, length):
-        '''
-        Read and return 'length' bytes from the virtual address 'vaddr'.
-        If any part of that block is unavailable, return None.
-        '''
-        return self._read_bytes(vaddr, length, pad = False)
-
     def is_valid_address(self, addr):
         vaddr = self.vtop(addr)
         return vaddr is not None and self.base.is_valid_address(vaddr)

volatility/args.py

@@ -24,6 +24,7 @@
 
 import argparse
 import logging
+import re
 import os
 import sys
 import zipfile
@@ -36,11 +37,22 @@
 class IntParser(argparse.Action):
     """Class to parse ints either in hex or as ints."""
     def parse_int(self, value):
+        # Support suffixes
+        multiplier = 1
+        m = re.search("(.*)(mb|kb|m|k)", value)
+        if m:
+            value = m.group(1)
+            suffix = m.group(2).lower()
+            if suffix in ("mb", "m"):
+                multiplier = 1024 * 1024
+            elif suffix in ("kb", "k"):
+                multiplier = 1024
+
         try:
             if value.startswith("0x"):
-                value = int(value, 16)
+                value = int(value, 16) * multiplier
             else:
-                value = int(value)
+                value = int(value) * multiplier
         except ValueError:
             raise argparse.ArgumentError(self, "Invalid integer value")
 

volatility/obj.py

@@ -270,12 +270,6 @@ def __getattr__(self, attr):
 
         return getattr(proxied, attr)
 
-    def __setattr__(self, attr, value):
-        try:
-            object.__setattr__(self, attr, value)
-        except AttributeError:
-            pass
-
     def __nonzero__(self):
         """ This method is called when we test the truth value of an
         Object. In volatility we consider an object to have True truth
@@ -447,6 +441,9 @@ def proxied(self, attr):
     def __radd__(self, other):
         return self.v() + long(other)
 
+    def __rsub__(self, other):
+        return self.v() - long(other)
+
     def size(self):
         return struct.calcsize(self.format_string)
 
@@ -571,7 +568,7 @@ def is_valid(self):
     def __getitem__(self, item):
         return self.dereference()[item]
 
-    def __setattr__(self, attr, value):
+    def __XXXXsetattr__(self, attr, value):
         if (attr in self.__dict__ or hasattr(self.__class__, attr) or
             not self._initialized):
             return super(Pointer, self).__setattr__(attr, value)
@@ -1043,7 +1040,7 @@ def __getattr__(self, attr):
 
         return self.m(attr)
 
-    def __setattr__(self, attr, value):
+    def __XXXsetattr__(self, attr, value):
         """Change underlying members"""
         # Special magic to allow initialization this test allows attributes to
         # be set in the __init__ method.

volatility/plugins/addrspaces/accelerated.py

@@ -33,18 +33,10 @@ def __init__(self, **kwargs):
         self._delegate = support.AMD64PagedMemory(self.base, int(self.dtb))
 
     def read(self, offset, length):
-        return self._delegate.read(offset, length)
+        return self._delegate.read(int(offset), int(length))
 
     def vtop(self, address):
-        return self._delegate.vtop(address)
-
-    def get_address_ranges(self, start=0, end=None):
-        if end is None:
-            end = 0xfffffffffffff
-
-        for virt, offset, _ in self._delegate.get_available_addresses(
-            start=start, length=end-start):
-            yield virt, offset
+        return self._delegate.vtop(int(address))
 
     def get_available_addresses(self):
         for virt, offset, _ in self._delegate.get_available_addresses():

volatility/plugins/addrspaces/intel.py

@@ -163,7 +163,7 @@ def vtop(self, vaddr):
         The function should return either None (no valid mapping)
         or the offset in physical memory where the address maps.
         '''
-        pde_value = self.get_pd(vaddr)
+        pde_value = self.get_pde(vaddr)
         if not self.entry_present(pde_value):
             # Add support for paged out PDE
             # (insert buffalo here!)

volatility/plugins/addrspaces/mmap_address_space.py

@@ -72,10 +72,11 @@ def __init__(self, filename=None, **kwargs):
             raise addrspace.ASAssertionError("Unable to mmap: %s" % e)
 
     def read(self, addr, length):
-        if addr == None:
-            return None
+        result = ""
+        if addr != None:
+            result = self.map[addr:addr+length]
 
-        return self.map[addr:addr+length]
+        return result + "\x00" * (length - len(result))
 
     def get_available_addresses(self):
         # TODO: Explain why this is always fsize - 1?

volatility/plugins/core.py

@@ -273,6 +273,34 @@ def __init__(self, pas_spec = "auto", **kwargs):
         super(LoadAddressSpace, self).__init__(**kwargs)
         self.pas_spec = pas_spec
 
+    def ResolveAddressSpace(self, name):
+        """Resolve the name into an address space.
+
+        This function is intended to be called from plugins which allow an
+        address space to be specified on the command line. We implement a simple
+        way for the user to specify the address space using a string. The
+        following formats are supported:
+
+        Kernel, K : Represents the kernel address space.
+        Physical, P: Represents the physical address space.
+
+        as_type@dtb_address: Instantiates the address space at the specified
+            DTB. For example: amd64@0x18700
+        """
+        # We can already specify a proper address space here.
+        if isinstance(name, addrspace.BaseAddressSpace):
+            return name
+
+        if name == "K" or name == "Kernel":
+            return (self.session.kernel_address_space or
+                    self.GetVirtualAddressSpace())
+
+        if name == "P" or name == "Physical":
+            return (self.session.physical_address_space or
+                    self.GetPhysicalAddressSpace())
+
+        return self.session.default_address_space
+
     def GetPhysicalAddressSpace(self):
         try:
             # Try to get a physical address space.
@@ -288,6 +316,8 @@ def GetPhysicalAddressSpace(self):
         except addrspace.ASAssertionError, e:
             logging.error("Could not create address space: %s" % e)
 
+        return self.session.physical_address_space
+
     def GetVirtualAddressSpace(self):
         if not self.session.physical_address_space:
             raise plugin.PluginError("Unable to find kernel address space.")
@@ -296,7 +326,6 @@ def GetVirtualAddressSpace(self):
         if self.profile is None:
             raise plugin.PluginError("Must specify a profile to load virtual AS.")
 
-
         address_space_curry = obj.Curry(
             GetAddressSpaceImplementation(self.profile),
             base=self.session.physical_address_space,
@@ -322,7 +351,7 @@ def GetVirtualAddressSpace(self):
 
             if not self.session.kernel_address_space:
                 raise plugin.PluginError(
-                    "A DTB value was found but failed to verifyo. "
+                    "A DTB value was found but failed to verify. "
                     "You can try setting it manualy using --dtb. "
                     "This could also happen when the profile is incorrect.")
 
@@ -333,6 +362,8 @@ def GetVirtualAddressSpace(self):
         if self.session.default_address_space is None:
             self.session.default_address_space = self.session.kernel_address_space
 
+        return self.session.kernel_address_space
+
     def GuessAddressSpace(self, base_as=None, **kwargs):
         """Loads an address space by stacking valid ASes on top of each other
         (priority order first).
@@ -658,13 +689,19 @@ class Grep(plugin.Command):
     @classmethod
     def args(cls, parser):
         super(Grep, cls).args(parser)
-        parser.add_argument("address_space", help="Name of the address_space to search.")
-        parser.add_argument("offset", default=0,  help="Start searching from this offset.")
+        parser.add_argument("address_space",
+                            help="Name of the address_space to search.")
+
+        parser.add_argument("offset", default=0,
+                            help="Start searching from this offset.")
+
         parser.add_argument("keyword", help="The binary string to find.")
-        parser.add_argument("limit", default=1024*1024, help="The length of data to search.")
 
-    def __init__(self, address_space=None, offset=0, keyword=None, context=20, limit=1024 * 1024,
-                 **kwargs):
+        parser.add_argument("limit", default=1024*1024,
+                            help="The length of data to search.")
+
+    def __init__(self, address_space=None, offset=0, keyword=None, context=20,
+                 limit=1024 * 1024, **kwargs):
         """Search an address space for keywords.
 
         Args: