gistfile1.py

/* 
Developed by Andrew Case ( atcuno@gmail.com  )
License: GPL2+
*/

import volatility.obj as obj
import volatility.commands as commands
import volatility.conf as conf
import volatility.utils as utils
import os,sys
config = conf.ConfObject()

def parse_system_map():

        smap = {}

        if not config.system_map:
                config.error("No system map file provided!")

        else:
                try:
                        os.stat(config.system_map)
                except:
                        config.error("Cannot open supplied system map file! %s" % sys.exc_info()[1])

                # get the system map
                for line in open(config.system_map,"r").readlines():

                        line = line.strip("\n")

                        (address,type,symbol) = line.split(" ")

                        smap[symbol] = int(address,16)

        return smap

def get_system_map():

        system_map = parse_system_map()

        return system_map

class linux_task_list_ps(commands.command):

        ''' gathers active tasks by walking the task_struct->task list '''

        def __init__(self, *args):
                commands.command.__init__(self, *args)

                self.addr_space = utils.load_as()

                self.smap = get_system_map()

        def offsetof(self, struct_name, list_member, profile):

                offset = profile.typeDict[struct_name][1][list_member][0]
                return offset

        # will be placed into a utility file ....
        # similar to for_each_process for this usage
        def walk_list_head(self, struct_name, list_member,list_head_ptr, addr_space):

                list_ptr = list_head_ptr.next

                while 1:

                        # return the address of the beginning of the strucutre, similar to list.h in kernel
                        yield list_ptr - self.offsetof(struct_name, list_member, addr_space.profile)

                        list_ptr = obj.Object("list_head", vm=addr_space, offset=list_ptr)

                        list_ptr = list_ptr.next

                        if list_ptr == list_head_ptr:
                                break

        def calculate(self):

                init_task_addr = self.smap["init_task"]

                init_task = obj.Object("task_struct", vm=self.addr_space, offset=init_task_addr)

                # walk the ->tasks list, note that this will *not* display "swapper"
                for task_addr in self.walk_list_head("task_struct", "tasks", init_task.tasks, self.addr_space):

                        task = obj.Object("task_struct", vm=self.addr_space, offset=task_addr)

                        yield task

        def render_text(self, outfd, data):

                print "%-20s %-15s %-15s" % ("Name","Pid","Uid")

                profile = self.addr_space.profile

                for task in data:
                        outfd.write("%-20s %-15d %-15d\n" % (task.comm, task.pid, profile.ts.get_uid(task)))