a k#d-?@sddlZddlZddlmZddlmZddlmZmZm Z m Z m Z m Z m Z mZddlmZddlmZddlmZddlmZdd lmZmZddlmmZdd lmZmZm Z dd l!m"Z"dd l#m$Z$m%Z%m&Z&m'Z'm(Z(m)Z)e*ej+Gd ddZ,dS)N)product)Event)AnyUnionListOptionalDictCallableTypecast)log_automotive)Graph)Scapy_Exception) SuperSocket)make_lined_tableSingleConversationSocket)EcuState EcuResponseEcu)'AutomotiveTestCaseExecutorConfiguration)AutomotiveTestCaseABC _SocketUnion_CleanupCallableStateGeneratorTestCaseGeneratorAutomotiveTestCasecseZdZdZeddZd1ddZfddZeej d d Z ed d Z ed dZ eddZ eddZddZddZd2ddZddZddZddZdd Zd!d"Zd3d#d$Zd%d&Zd'd(Zd)d*Zd+d,Zd-d.Zed/d0ZZS)4AutomotiveTestCaseExecutora Base class for different automotive scanners. This class handles the connection to a scan target, ensures the execution of all it's test cases, and stores the system state machine :param socket: A socket object to communicate with the scan target :param reset_handler: A function to reset the scan target :param reconnect_handler: In case the communication needs to be established after a reset, provide a reconnect function which returns a socket object :param test_cases: A list of TestCase instances or classes :param kwargs: Arguments for the internal AutomotiveTestCaseExecutorConfiguration instance cCs tddS)N)session)rselfr!`/var/www/html/gsl.tools/lib/python3.9/site-packages/scapy/contrib/automotive/scanner/executor.py_initial_ecu_state1sz-AutomotiveTestCaseExecutor._initial_ecu_stateNcKsdt|tst||_n||_|j|_||_||_t|_t |pB|j fi||_ | t |_dSN) isinstancersocketr# target_state reset_handlerreconnect_handlerlistcleanup_functionsrdefault_test_case_clss configurationvalidate_test_case_kwargsr_stop_scan_event)r r&r(r) test_caseskwargsr!r!r"__init__6s  z#AutomotiveTestCaseExecutor.__init__cstt|\}}}z |d=Wnty0Yn0z |d=WntyNYn0z |d=WntylYn0z |d=WntyYn0|||fS)Nr&r(r)r/)superr __reduce__KeyError)r ftd __class__r!r"r4Ss$        z%AutomotiveTestCaseExecutor.__reduce__cCs tdSr$)NotImplementedErrorrr!r!r"r,gsz1AutomotiveTestCaseExecutor.default_test_case_clsscCs|jjSr$)r- state_graphrr!r!r"r<msz&AutomotiveTestCaseExecutor.state_graphcs:fddjjD}tdd|DjggdddS)z Returns all state paths. A path is represented by a list of EcuState objects. :return: A list of paths. cs(g|] }|jkrtjj|qSr!)r#r Zdijkstrar<).0srr!r" zs z:AutomotiveTestCaseExecutor.state_paths..cSsg|] }|r|qSr!r!r=pr!r!r"r?~cSs|dS)Nr!)xr!r!r"rBz8AutomotiveTestCaseExecutor.state_paths..key)r<nodessortedr#)r pathsr!rr" state_pathsrs z&AutomotiveTestCaseExecutor.state_pathscCsdd|jDS)z{ Returns a list with all final states. A final state is the last state of a path. :return: cSsg|] }|dqS)rCr!r@r!r!r"r?rBz;AutomotiveTestCaseExecutor.final_states..)rKrr!r!r" final_statessz'AutomotiveTestCaseExecutor.final_statescCstddt|jj|jDS)Ncss|]\}}||VqdSr$) has_completed)r=r7r>r!r!r" rBz.)allrr-r0rLrr!r!r"scan_completedsz)AutomotiveTestCaseExecutor.scan_completedcCs$td|jr||j|_dS)Nz Target reset)r infor(r#r'rr!r!r" reset_targets z'AutomotiveTestCaseExecutor.reset_targetc Cs|jrtz|jWn0tyD}ztd|WYd}~n d}~00td|}t|tsnt||_n||_|jj rt ddS)Nz"Exception '%s' during socket.closezTarget reconnectz.Socket closed even after reconnect. Stop scan!) r)r&close Exceptionr exceptionrQr%rclosedr)r er&r!r!r" reconnects    z$AutomotiveTestCaseExecutor.reconnectcCs||j|j|jz|j|jj}Wnty>t}Yn0|rttt |t d}| dd}t |||d<t d|jj||j|j|jfd|ji|||j|j|j||||t|dr |}|dur t d|jj|d |d |d dS) a This function ensures the correct execution of a testcase, including the pre_execute, execute and post_execute. Finally the testcase is asked if a new edge or a new testcase was generated. :param test_case: A test case to be executed :param kill_time: If set, this defines the maximum execution time for the current test_case :return: None Zexecution_timeiz!Execute test_case %s with args %sZ stop_eventruntime_estimationNz/[i] Test_case %s: TODO %d, DONE %d, TOTAL %0.2frr)Z pre_executer&r'r-r:__name__r5dictmaxinttimegetminr debugexecuter/Z post_executecheck_new_statescheck_new_testcaseshasattrrZ)r test_case kill_timetest_case_kwargsZmax_execution_timeZcur_execution_timeZ estimationr!r!r"execute_test_casesD           z,AutomotiveTestCaseExecutor.execute_test_casecCs2t|tr.|}|r.td||j|dS)NzTestcase generated %s)r%rZget_generated_test_caser rcr-Z add_test_case)r rhZ new_test_caser!r!r"rfs   z.AutomotiveTestCaseExecutor.check_new_testcasescCsJt|trF||j|j}|rFtd|||j|}|j ||dS)Nz Edge found %s) r%rZ get_new_edger&r-r rcZget_transition_functionr<Zadd_edge)r rhedgetfr!r!r"res   z+AutomotiveTestCaseExecutor.check_new_statescCs4|jjD]&}t|tr|j|jj}||qdSr$)r-r0r%rr:r\Z check_kwargs)r rhrjr!r!r"r.s  z4AutomotiveTestCaseExecutor.validate_test_case_kwargscCs|jdSr$)r/setrr!r!r" stop_scansz$AutomotiveTestCaseExecutor.stop_scancCsZg}|jjD]0}t|dsq |}|dur.q ||dq t|rVt|t|SdS)NrZr[g)r-r0rgrZappendlensum)r progresstcZestr!r!r"rss  z#AutomotiveTestCaseExecutor.progressc Cs|jt|pd}tdt||tkrd}td|td|jt |j|j j D]X\}}td||tk}|s|j rtdq|d}| |rtd t||qnzzJ||std |WW|qntd t|||||d }Wntttfy}zptd ||j jrR|t|trrtd|jtt|jjr|jdurtd|WYd}~n d}~00W|qn|0qn|s.tdqq.|| dS)z| Executes all testcases for a given time. :param timeout: Time for execution. :return: None lzSet kill_time to %sFz[i] Scan progress %0.2fz[i] Scan paths %sz Scan path %sz*Execution time exceeded. Terminating scan!rCzState %s for %s completedzError entering path %szExecute %s for path %sTz Exception: %sz OSError occurred, closing socketNz$Socket went down. Need to leave scanz-Execute failure or scan completed. Exit scan!)!r/clearr`r rcctimerQrsrKrr-r0is_setrMreprenter_state_patherror cleanup_statestrrkOSError ValueErrorrrUr%r&rSr rrVr)criticalrR) r timeoutriZtest_case_executedrArh terminateZ final_staterWr!r!r"scansx            zAutomotiveTestCaseExecutor.scancCsz|d|jkrtd||t|dkr6dS|ddD]2}|j|f}|j|sB|j|| dSqBdS)z Resets and reconnects to a target and applies all transition functions to traversal a given path. :param path: Path to be applied to the scan target. :return: True, if all transition functions could be executed. rz9Initial state of path not equal reset state of the targetrTNF) r#rrRrXrqr' enter_stater<Z downrate_edger{)r path next_staterlr!r!r"ryDs    z+AutomotiveTestCaseExecutor.enter_state_pathc Cs||f}|j|}|dur,td|dS|\}}}||j|j|}|rl||_|durh|j|g7_dStd|dSdS)aJ Obtains a transition function from the system state graph and executes it. On success, the cleanup function is added for a later cleanup of the new state. :param prev_state: Current state :param next_state: Desired state :return: True, if state could be changed successful NzNo transition function for %sFTzTransition for edge %s failed) r<Zget_transition_tuple_for_edger rzr&r-r'r+rQ) r Z prev_staterrlfuncsZ trans_funcZ trans_kwargs clean_funcZ state_changedr!r!r"r^s      z&AutomotiveTestCaseExecutor.enter_statec Cs||jD]h}t|sqz"||j|js4tdt|Wqttt fyl}zt d|WYd}~qd}~00qt |_dS)zf Executes all collected cleanup functions from a traversed path :return: None zCleanup function %s failedzException during cleanup: %sN) r+callabler&r-r rQrxr}r~rrr*)r r6rWr!r!r"r{|s $z(AutomotiveTestCaseExecutor.cleanup_statecCs|jjD] }|qdSr$)r-r0show)r r7r!r!r"show_testcasess z)AutomotiveTestCaseExecutor.show_testcasescCsRt}|jjD]0}|jjD]"}|t||jj||fg7}qqt |dddS)NcSs|d|d|dfS)Nrrr[r!)tupr!r!r"rErBzBAutomotiveTestCaseExecutor.show_testcases_status..) r*r-r0r<rHrxr:r\rMr)r datar7r>r!r!r"show_testcases_statuss   "z0AutomotiveTestCaseExecutor.show_testcases_statuscCs0t}|jjD]}||j7}q|jtjd|S)ac Returns a sorted list of supported responses, gathered from all enumerators. The sort is done in a way to provide the best possible results, if this list of supported responses is used to simulate an real world Ecu with the EcuAnsweringMachine object. :return: A sorted list of EcuResponse objects rF)r*r-r0supported_responsessortrZ sort_key_func)r rrtr!r!r"rs   z.AutomotiveTestCaseExecutor.supported_responses)NNN)N)N)r\ __module__ __qualname____doc__propertyr#r2r4abcabstractmethodr,r<rKrLrPrRrXrkrfrer.rorsrryrr{rrr __classcell__r!r!r9r"rsF         1  >r)-rr` itertoolsr threadingrZ scapy.compatrrrrrr r r Zscapy.contrib.automotiver Z&scapy.contrib.automotive.scanner.graphr Z scapy.errorrZscapy.supersocketrZ scapy.utilsrrZscapy.libs.sixZlibssixZscapy.contrib.automotive.ecurrrZ.scapy.contrib.automotive.scanner.configurationrZ*scapy.contrib.automotive.scanner.test_caserrrrrr add_metaclassABCMetarr!r!r!r" s  (