<s> """<STR_LIT>""" <EOL> import collections <EOL> import os <EOL> import StringIO <EOL> import yaml <EOL> from grr . lib import config_lib <EOL> from grr . lib import registry <EOL> from grr . lib import test_lib <EOL> from grr . lib import type_info <EOL> from grr . lib . checks import checks <EOL> from grr . lib . checks import filters <EOL> from grr . lib . checks import hints <EOL> from grr . lib . rdfvalues import anomaly as rdf_anomaly <EOL> from grr . lib . rdfvalues import client as rdf_client <EOL> from grr . lib . rdfvalues import paths as rdf_paths <EOL> from grr . parsers import linux_service_parser <EOL> class HostCheckTest ( test_lib . GRRBaseTest ) : <EOL> """<STR_LIT>""" <EOL> __metaclass__ = registry . MetaclassRegistry <EOL> loaded_checks = None <EOL> @ classmethod <EOL> def LoadCheck ( cls , cfg_file , * check_ids ) : <EOL> """<STR_LIT>""" <EOL> if HostCheckTest . loaded_checks is None : <EOL> HostCheckTest . loaded_checks = { } <EOL> cfg = os . path . join ( config_lib . CONFIG [ "<STR_LIT>" ] , "<STR_LIT>" , "<STR_LIT>" , <EOL> cfg_file ) <EOL> if check_ids : <EOL> key = "<STR_LIT>" % ( cfg , "<STR_LIT:U+002C>" . join ( check_ids ) ) <EOL> if key in HostCheckTest . loaded_checks : <EOL> return HostCheckTest . loaded_checks [ key ] <EOL> loaded = [ ] <EOL> for chk_id in check_ids : <EOL> loaded . append ( checks . LoadCheckFromFile ( cfg , chk_id ) ) <EOL> HostCheckTest . loaded_checks [ key ] = loaded <EOL> return loaded <EOL> else : <EOL> key = "<STR_LIT>" % cfg_file <EOL> if key in HostCheckTest . loaded_checks : <EOL> return HostCheckTest . loaded_checks [ key ] <EOL> else : <EOL> result = checks . LoadChecksFromFiles ( [ cfg ] ) <EOL> HostCheckTest . loaded_checks [ key ] = result <EOL> return result <EOL> def TestDataPath ( self , file_name ) : <EOL> """<STR_LIT>""" <EOL> path = os . path . join ( config_lib . CONFIG [ "<STR_LIT>" ] , file_name ) <EOL> if not os . path . isfile ( path ) : <EOL> raise test_lib . Error ( "<STR_LIT>" % file_name ) <EOL> return path <EOL> @ classmethod <EOL> def SetKnowledgeBase ( cls , hostname = "<STR_LIT>" , host_os = "<STR_LIT>" , <EOL> host_data = None ) : <EOL> """<STR_LIT>""" <EOL> if not host_data : <EOL> host_data = { } <EOL> kb = rdf_client . KnowledgeBase ( ) <EOL> kb . hostname = hostname <EOL> kb . os = host_os <EOL> host_data [ "<STR_LIT>" ] = kb <EOL> return host_data <EOL> def SetArtifactData ( self , anomaly = None , parsed = None , raw = None , results = None ) : <EOL> """<STR_LIT>""" <EOL> if not results : <EOL> results = { "<STR_LIT>" : [ ] , "<STR_LIT>" : [ ] , "<STR_LIT>" : [ ] } <EOL> results [ "<STR_LIT>" ] . extend ( anomaly or [ ] ) <EOL> results [ "<STR_LIT>" ] . extend ( parsed or [ ] ) <EOL> results [ "<STR_LIT>" ] . extend ( raw or [ ] ) <EOL> return results <EOL> def AddData ( self , parser , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return [ parser ( ) . Parse ( * args , ** kwargs ) ] <EOL> def AddListener ( self , ip , port , family = "<STR_LIT>" , sock_type = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> conn = rdf_client . NetworkConnection ( ) <EOL> conn . state = "<STR_LIT>" <EOL> conn . family = family <EOL> conn . type = sock_type <EOL> conn . local_address = rdf_client . NetworkEndpoint ( ip = ip , port = port ) <EOL> return conn <EOL> def RunChecks ( self , host_data , labels = None , restrict_checks = None ) : <EOL> """<STR_LIT>""" <EOL> return { r . check_id : r for r in checks . CheckHost ( <EOL> host_data , labels = labels , restrict_checks = restrict_checks ) } <EOL> def GetCheckErrors ( self , check_spec ) : <EOL> """<STR_LIT>""" <EOL> errors = [ ] <EOL> try : <EOL> check = checks . Check ( ** check_spec ) <EOL> check . Validate ( ) <EOL> except ( checks . Error , filters . Error , hints . Error , type_info . Error ) as e : <EOL> errors . append ( str ( e ) ) <EOL> except Exception as e : <EOL> errors . append ( "<STR_LIT>" % ( type ( e ) , e ) ) <EOL> return errors <EOL> def CreateStat ( self , path , uid = <NUM_LIT:0> , gid = <NUM_LIT:0> , mode = <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> pathspec = rdf_paths . PathSpec ( path = path , pathtype = "<STR_LIT>" ) <EOL> return rdf_client . StatEntry ( pathspec = pathspec , st_uid = uid , st_gid = gid , <EOL> st_mode = mode ) <EOL> def _AddToHostData ( self , host_data , artifact , data , parser ) : <EOL> """<STR_LIT>""" <EOL> if type ( data ) != dict : <EOL> raise test_lib . Error ( "<STR_LIT>" % artifact ) <EOL> rdfs = [ ] <EOL> stats = [ ] <EOL> files = [ ] <EOL> for path , lines in data . items ( ) : <EOL> stat = self . CreateStat ( path ) <EOL> stats . append ( stat ) <EOL> file_obj = StringIO . StringIO ( lines ) <EOL> files . append ( file_obj ) <EOL> if not parser . process_together : <EOL> rdfs . extend ( list ( parser . Parse ( stat , file_obj , None ) ) ) <EOL> if parser . process_together : <EOL> rdfs = list ( parser . ParseMultiple ( stats , files , None ) ) <EOL> host_data [ artifact ] = self . SetArtifactData ( <EOL> anomaly = [ a for a in rdfs if isinstance ( a , rdf_anomaly . Anomaly ) ] , <EOL> parsed = [ r for r in rdfs if not isinstance ( r , rdf_anomaly . Anomaly ) ] , <EOL> raw = stats , results = host_data . get ( artifact ) ) <EOL> return host_data <EOL> def GenResults ( self , artifact_list , sources_list , parser_list = None ) : <EOL> """<STR_LIT>""" <EOL> if parser_list is None : <EOL> parser_list = [ None ] * len ( artifact_list ) <EOL> if any ( type ( lst ) != list for lst in [ artifact_list , sources_list , <EOL> parser_list ] ) : <EOL> raise test_lib . Error ( "<STR_LIT>" ) <EOL> if any ( len ( lst ) != len ( artifact_list ) for lst in [ sources_list , <EOL> parser_list ] ) : <EOL> raise test_lib . Error ( "<STR_LIT>" ) <EOL> host_data = self . SetKnowledgeBase ( ) <EOL> for artifact , sources , parser in zip ( artifact_list , sources_list , <EOL> parser_list ) : <EOL> if parser is None : <EOL> host_data [ artifact ] = self . SetArtifactData ( <EOL> raw = sources , results = host_data . get ( artifact ) ) <EOL> else : <EOL> host_data = self . _AddToHostData ( host_data , artifact , sources , parser ) <EOL> return self . RunChecks ( host_data ) <EOL> def GenProcessData ( self , processes ) : <EOL> """<STR_LIT>""" <EOL> host_data = self . SetKnowledgeBase ( ) <EOL> data = [ ] <EOL> for ( name , pid , cmdline ) in processes : <EOL> data . append ( rdf_client . Process ( name = name , pid = pid , cmdline = cmdline ) ) <EOL> host_data [ "<STR_LIT>" ] = self . SetArtifactData ( raw = data ) <EOL> return host_data <EOL> @ classmethod <EOL> def _GenFileData ( cls , paths , data , stats = None , files = None , modes = None ) : <EOL> """<STR_LIT>""" <EOL> if stats is None : <EOL> stats = [ ] <EOL> if files is None : <EOL> files = [ ] <EOL> if modes is None : <EOL> modes = { } <EOL> modes . setdefault ( "<STR_LIT>" , <NUM_LIT:0> ) <EOL> modes . setdefault ( "<STR_LIT>" , <NUM_LIT:0> ) <EOL> modes . setdefault ( "<STR_LIT>" , <NUM_LIT> ) <EOL> for path in paths : <EOL> p = rdf_paths . PathSpec ( path = path , pathtype = "<STR_LIT>" ) <EOL> stats . append ( rdf_client . StatEntry ( pathspec = p , ** modes ) ) <EOL> for val in data : <EOL> files . append ( StringIO . StringIO ( val ) ) <EOL> return stats , files <EOL> @ classmethod <EOL> def GenStatFileData ( cls , data , modes = None ) : <EOL> """<STR_LIT>""" <EOL> paths = [ ] <EOL> contents = [ ] <EOL> for path , content in data . iteritems ( ) : <EOL> paths . append ( path ) <EOL> contents . append ( content ) <EOL> return cls . _GenFileData ( paths , contents , modes = modes ) <EOL> def GenFileData ( self , artifact , data , parser = None , modes = None , include = None ) : <EOL> """<STR_LIT>""" <EOL> host_data = self . SetKnowledgeBase ( ) <EOL> if not parser : <EOL> raise test_lib . Error ( "<STR_LIT>" ) <EOL> if not modes : <EOL> modes = { } <EOL> kb = host_data [ "<STR_LIT>" ] <EOL> files = [ ] <EOL> rdfs = [ ] <EOL> stats = [ ] <EOL> for path in data : <EOL> if include and path not in include : <EOL> continue <EOL> file_modes = modes . get ( path , { } ) <EOL> stats , files = self . _GenFileData ( [ path ] , [ data [ path ] ] , stats = stats , <EOL> files = files , modes = file_modes ) <EOL> if parser . process_together : <EOL> rdfs = list ( parser . ParseMultiple ( stats , files , kb ) ) <EOL> else : <EOL> for stat , file_obj in zip ( stats , files ) : <EOL> rslt = parser . Parse ( stat , file_obj , kb ) or [ ] <EOL> rdfs . extend ( rslt ) <EOL> anomaly = [ a for a in rdfs if isinstance ( a , rdf_anomaly . Anomaly ) ] <EOL> parsed = [ r for r in rdfs if not isinstance ( r , rdf_anomaly . Anomaly ) ] <EOL> host_data [ artifact ] = self . SetArtifactData ( <EOL> parsed = parsed , anomaly = anomaly , raw = stats ) <EOL> return host_data <EOL> def GenSysVInitData ( self , links ) : <EOL> """<STR_LIT>""" <EOL> return self . GenFileData ( artifact = "<STR_LIT>" , <EOL> data = { x : "<STR_LIT>" for x in links } , <EOL> parser = linux_service_parser . LinuxSysVInitParser ( ) , <EOL> modes = { x : { "<STR_LIT>" : <NUM_LIT> } for x in links } ) <EOL> def assertRanChecks ( self , check_ids , results ) : <EOL> """<STR_LIT>""" <EOL> self . assertTrue ( set ( check_ids ) . issubset ( set ( results . keys ( ) ) ) ) <EOL> def assertChecksNotRun ( self , check_ids , results ) : <EOL> """<STR_LIT>""" <EOL> self . assertFalse ( set ( check_ids ) . intersection ( set ( results . keys ( ) ) ) ) <EOL> def assertResultEqual ( self , rslt1 , rslt2 ) : <EOL> """<STR_LIT>""" <EOL> if rslt1 . check_id != rslt2 . check_id : <EOL> self . fail ( "<STR_LIT>" % ( rslt1 . check_id , rslt2 . check_id ) ) <EOL> rslt1_anoms = { } <EOL> for a in rslt1 . anomaly : <EOL> anoms = rslt1_anoms . setdefault ( a . symptom , [ ] ) <EOL> anoms . extend ( a . finding ) <EOL> rslt2_anoms = { } <EOL> for a in rslt2 . anomaly : <EOL> anoms = rslt2_anoms . setdefault ( a . symptom , [ ] ) <EOL> anoms . extend ( a . finding ) <EOL> self . assertItemsEqual ( rslt1_anoms , rslt2_anoms , <EOL> "<STR_LIT>" % <EOL> ( rslt1_anoms . keys ( ) , rslt2_anoms . keys ( ) ) ) <EOL> for symptom , findings in rslt1_anoms . iteritems ( ) : <EOL> rslt1_found = [ f . strip ( ) for f in findings ] <EOL> rslt2_found = [ f . strip ( ) for f in rslt2_anoms [ symptom ] ] <EOL> self . assertItemsEqual ( rslt1_found , rslt2_found ) <EOL> def assertIsCheckIdResult ( self , rslt , expected ) : <EOL> """<STR_LIT>""" <EOL> self . assertIsInstance ( rslt , checks . CheckResult ) <EOL> self . assertEqual ( expected , rslt . check_id ) <EOL> def assertValidCheck ( self , check_spec ) : <EOL> """<STR_LIT>""" <EOL> errors = self . GetCheckErrors ( check_spec ) <EOL> if errors : <EOL> self . fail ( "<STR_LIT:\n>" . join ( errors ) ) <EOL> def assertValidCheckFile ( self , path ) : <EOL> """<STR_LIT>""" <EOL> prefix = os . path . commonprefix ( config_lib . CONFIG [ "<STR_LIT>" ] ) <EOL> relpath = os . path . relpath ( path , prefix ) <EOL> try : <EOL> configs = checks . LoadConfigsFromFile ( path ) <EOL> except yaml . error . YAMLError as e : <EOL> self . fail ( "<STR_LIT>" % ( relpath , e ) ) <EOL> errors = collections . OrderedDict ( ) <EOL> for check_id , check_spec in configs . iteritems ( ) : <EOL> check_errors = self . GetCheckErrors ( check_spec ) <EOL> if check_errors : <EOL> msg = errors . setdefault ( relpath , [ "<STR_LIT>" % check_id ] ) <EOL> msg . append ( check_errors ) <EOL> if errors : <EOL> message = "<STR_LIT>" <EOL> for k , v in errors . iteritems ( ) : <EOL> message += "<STR_LIT>" % k <EOL> message += "<STR_LIT>" % v [ <NUM_LIT:0> ] <EOL> for err in v [ <NUM_LIT:1> ] : <EOL> message += "<STR_LIT>" % err <EOL> self . fail ( message ) <EOL> def _HasSymptom ( self , anomalies , sym ) : <EOL> """<STR_LIT>""" <EOL> if sym is None : <EOL> return True <EOL> rslts = { rslt . symptom : rslt for rslt in anomalies } <EOL> rslt = rslts . get ( sym ) <EOL> self . assertTrue ( rslt is not None , <EOL> "<STR_LIT>" % <EOL> ( sym , "<STR_LIT:U+002C>" . join ( rslts ) ) ) <EOL> def _GetFindings ( self , anomalies , sym ) : <EOL> """<STR_LIT>""" <EOL> result = set ( ) <EOL> for anomaly in anomalies : <EOL> if anomaly . symptom == sym : <EOL> result . update ( set ( anomaly . finding ) ) <EOL> return result <EOL> def _MatchFindings ( self , expected , found ) : <EOL> """<STR_LIT>""" <EOL> matched_so_far = set ( ) <EOL> for finding_str in expected : <EOL> no_match = True <EOL> for found_str in found : <EOL> if finding_str in found_str : <EOL> matched_so_far . add ( found_str ) <EOL> no_match = False <EOL> break <EOL> if no_match : <EOL> return False <EOL> if not matched_so_far . symmetric_difference ( found ) : <EOL> return True <EOL> def assertCheckDetectedAnom ( self , check_id , results , sym = None , findings = None ) : <EOL> """<STR_LIT>""" <EOL> chk = results . get ( check_id ) <EOL> self . assertTrue ( chk is not None , "<STR_LIT>" % check_id ) <EOL> self . assertTrue ( chk , "<STR_LIT>" % check_id ) <EOL> self . _HasSymptom ( chk . anomaly , sym ) <EOL> if findings is None : <EOL> return True <EOL> findings = set ( findings ) <EOL> found = self . _GetFindings ( chk . anomaly , sym ) <EOL> if self . _MatchFindings ( findings , found ) : <EOL> return True <EOL> others = "<STR_LIT:\n>" . join ( [ str ( a ) for a in chk . anomaly ] ) <EOL> self . fail ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> % ( sym , "<STR_LIT:U+002CU+0020>" . join ( findings ) , "<STR_LIT:U+002CU+0020>" . join ( found ) , others ) ) <EOL> def assertCheckUndetected ( self , check_id , results ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( results , collections . Mapping ) : <EOL> self . fail ( "<STR_LIT>" % type ( results ) ) <EOL> if check_id not in results : <EOL> self . fail ( "<STR_LIT>" % check_id ) <EOL> if results . get ( check_id ) : <EOL> self . fail ( "<STR_LIT>" <EOL> % ( check_id , results . get ( check_id ) . anomaly ) ) <EOL> def assertChecksUndetected ( self , check_ids , results ) : <EOL> """<STR_LIT>""" <EOL> for check_id in check_ids : <EOL> self . assertCheckUndetected ( check_id , results ) </s>
