<s> '''<STR_LIT>''' <EOL> import os <EOL> import hashlib <EOL> import pprint <EOL> import collections <EOL> from rekall_adapter . rekall_adapter import RekallAdapter <EOL> class MemoryImageConnScan ( object ) : <EOL> '''<STR_LIT>''' <EOL> dependencies = [ '<STR_LIT>' ] <EOL> def __init__ ( self ) : <EOL> '''<STR_LIT>''' <EOL> self . plugin_name = '<STR_LIT>' <EOL> self . current_table_name = '<STR_LIT>' <EOL> self . output = { '<STR_LIT>' : collections . defaultdict ( list ) } <EOL> self . column_map = { } <EOL> def execute ( self , input_data ) : <EOL> '''<STR_LIT>''' <EOL> adapter = RekallAdapter ( ) <EOL> adapter . set_plugin_name ( self . plugin_name ) <EOL> rekall_output = adapter . execute ( input_data ) <EOL> for line in rekall_output : <EOL> if line [ '<STR_LIT:type>' ] == '<STR_LIT:m>' : <EOL> self . output [ '<STR_LIT>' ] = line [ '<STR_LIT:data>' ] <EOL> elif line [ '<STR_LIT:type>' ] == '<STR_LIT:s>' : <EOL> self . current_table_name = line [ '<STR_LIT:data>' ] [ '<STR_LIT:name>' ] [ <NUM_LIT:1> ] <EOL> elif line [ '<STR_LIT:type>' ] == '<STR_LIT:t>' : <EOL> self . column_map = { item [ '<STR_LIT>' ] : item [ '<STR_LIT:name>' ] if '<STR_LIT:name>' in item else item [ '<STR_LIT>' ] for item in line [ '<STR_LIT:data>' ] } <EOL> elif line [ '<STR_LIT:type>' ] == '<STR_LIT:r>' : <EOL> row = RekallAdapter . process_row ( line [ '<STR_LIT:data>' ] , self . column_map ) <EOL> self . output [ '<STR_LIT>' ] [ self . current_table_name ] . append ( row ) <EOL> else : <EOL> print '<STR_LIT>' % ( line [ '<STR_LIT:type>' ] , line [ '<STR_LIT:data>' ] ) <EOL> return self . output <EOL> import pytest <EOL> @ pytest . mark . xfail <EOL> def test ( ) : <EOL> '''<STR_LIT>''' <EOL> import zerorpc <EOL> workbench = zerorpc . Client ( timeout = <NUM_LIT> , heartbeat = <NUM_LIT> ) <EOL> workbench . connect ( "<STR_LIT>" ) <EOL> data_path = os . path . join ( os . path . dirname ( os . path . realpath ( __file__ ) ) , '<STR_LIT>' ) <EOL> if not os . path . isfile ( data_path ) : <EOL> print '<STR_LIT>' <EOL> import urllib <EOL> urllib . urlretrieve ( '<STR_LIT>' , data_path ) <EOL> if not os . path . isfile ( data_path ) : <EOL> print '<STR_LIT>' <EOL> print '<STR_LIT>' <EOL> exit ( <NUM_LIT:1> ) <EOL> if os . stat ( data_path ) . st_size < <NUM_LIT> : <EOL> data_path = os . path . join ( os . path . dirname ( os . path . realpath ( __file__ ) ) , '<STR_LIT>' ) <EOL> with open ( data_path , '<STR_LIT:rb>' ) as mem_file : <EOL> print '<STR_LIT>' % mem_file . read ( ) [ : <NUM_LIT> ] <EOL> print '<STR_LIT>' <EOL> print '<STR_LIT>' <EOL> exit ( <NUM_LIT:1> ) <EOL> data_path = os . path . join ( os . path . dirname ( os . path . realpath ( __file__ ) ) , '<STR_LIT>' ) <EOL> with open ( data_path , '<STR_LIT:rb>' ) as mem_file : <EOL> raw_bytes = mem_file . read ( ) <EOL> md5 = hashlib . md5 ( raw_bytes ) . hexdigest ( ) <EOL> if not workbench . has_sample ( md5 ) : <EOL> md5 = workbench . store_sample ( open ( data_path , '<STR_LIT:rb>' ) . read ( ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> worker = MemoryImageConnScan ( ) <EOL> output = worker . execute ( { '<STR_LIT>' : { '<STR_LIT>' : raw_bytes } } ) <EOL> print '<STR_LIT>' <EOL> print '<STR_LIT>' % output [ '<STR_LIT>' ] <EOL> for name , table in output [ '<STR_LIT>' ] . iteritems ( ) : <EOL> print '<STR_LIT>' % name <EOL> pprint . pprint ( table ) <EOL> assert '<STR_LIT>' not in output <EOL> output = workbench . work_request ( '<STR_LIT>' , md5 ) [ '<STR_LIT>' ] <EOL> print '<STR_LIT>' <EOL> print '<STR_LIT>' % output [ '<STR_LIT>' ] <EOL> for name , table in output [ '<STR_LIT>' ] . iteritems ( ) : <EOL> print '<STR_LIT>' % name <EOL> pprint . pprint ( table ) <EOL> assert '<STR_LIT>' not in output <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> test ( ) </s>
