hhpddlZddlZddlZddlZddlmZddlmZddlm Z ddl m Z ddl m Z mZddlmZddlmZdd lmZdd lmZdd lmZdd lmZmZmZmZmZmZm Z m!Z!m"Z"m#Z#gd Z$Gdde%Z&Gdde%Z'GddeZ(GddZ)GddeeZ*dS)N) defaultdict) GeneratorType)Lock)orb)RawPacket) PacketList)DefaultSession)AnsweringMachine) SuperSocket)Scapy_Exception) AnyUnionIterableCallableListOptionalTupleTypecastDict)EcuStateEcu EcuResponse EcuSessionEcuAnsweringMachineceZdZdZddgZdZdZedZdZ dZ d Z d Z d Z d Zd ZdZdZdZdZdZedZedZeddZdS)rz Stores the state of an Ecu. The state is defined by a protocol, for example UDS or GMLAN. A EcuState supports comparison and serialization (command()). __dict__ __cache__c d|_|D]?\}}t|trt |}|||@dSN)ritems isinstancerlist __setitem__)selfkwargskvs `/mounts/lovelace/software/anaconda3/lib/python3.11/site-packages/scapy/contrib/automotive/ecu.py__init__zEcuState.__init__5sdLLNN # #DAq!]++ GG   Q " " " " # #ctj}tj}jjd|kr~t}t jfd|DD]L}i}t|D]\}}||||||<|tdi|M||f_jdS)Nc:g|]}|S)_flatten).0r)r&s r* z$EcuState._expand..Cs%(J(J(Jaq)9)9(J(J(Jr,rr0) r$rvalueskeysr itertoolsproduct enumerateappendr)r&r4r5expandedxr'ir(s` r*_expandzEcuState._expand=sdm**,,--DM&&(()) > !T^A%6&%@%@vvH&(J(J(J(J6(J(J(JK 4 4%dOO%%DAqt| !!F1II 2 26 2 23333&/DN~a  r,cpt|ttfr|gSt|dr,t|drt |dkr t |St|ds|gSt }|D]6}t|dr|t |z }0||gz }7|S)N__iter____len__r.)r#strbyteshasattrlenr$rr1)r; flattenedys r*r1zEcuState._flattenMs a#u & & 3J Q # # 9(=(= #a&&A++8OJ'' 3JFF  ! !Aq*%% !X..q111 aS r,c$d|_|j|=dSr!rr)r&keys r* __delitem__zEcuState.__delitem__^s M#   r,cNt|jSr!)rDrr5r&s r*r@zEcuState.__len__cs4=%%''(((r,c|j|Sr!rr&items r* __getitem__zEcuState.__getitem__gs}T""r,c(d|_||j|<dSr!rH)r&rIvalues r*r%zEcuState.__setitem__ks" cr,cddt|jdDS)Nc3ZK|]&\}}t|t|zV'dSr!)rAr2r(r)s r* z$EcuState.__repr__..rsNJJ41as1vvAJJJJJJr,c|dSNrr0ts r*z#EcuState.__repr__..ss 1Q4r,rIjoinsortedrr"rLs r*__repr__zEcuState.__repr__psYwwJJdm1133HHHJJJJJ Jr,ctttjtjkrdS t fdjDS#t $rYdSwxYw)NFc3PK|] }j|j|kV!dSr!rN)r2r(otherr&s r*rXz"EcuState.__eq__..{sJ66}Q'5>!+<<666666r,)rrrDrallr5KeyErrorr&res``r*__eq__zEcuState.__eq__usXu%% t}  U^!4!4 4 45 66666 $ 2 2 4 466666 6   55 s2A88 BBct|tsdStfd|DS)NFc3DK|]}|vVdSr!)r=)r2sr&s r*rXz(EcuState.__contains__..s0??11 &??????r,)r#rrfr=rOs` r* __contains__zEcuState.__contains__sE$)) 5???? ??????r,c||k Sr!r0rhs r*__ne__zEcuState.__ne__sD=  r,c (||krdSt|t|krdSt|t|krdSt|jt|j}t |D]}t |j|t|j|sHtdt|j|dt|j|d||j||j|krdS|j||j|krdSt|t|jkrt|j t|j}t|j t|j}t||D]\}}||krdSdStd|jd|j)NFTzCan't compare z with z for the EcuState element z;EcuStates should be identical. Something bad happen. self: z other: ) rDsetrr5 intersectionrar#type TypeError differencezip)r&recommonr( self_diffs other_diffsrlos r*__lt__zEcuState.__lt__so 5==5 t99s5zz ! !4 t99s5zz ! !5T]''))**77 ##%% & &((  AenQ/dmA6F1G1GHH Ji$-*++++T%.2C-D-D-D-DaaIJJJ}Q%."333tt}Q%."333uu4 v;;T]++ + +T]//1122==EN''))**,,Jen113344??DM&&(())++KJ 44  1q55445i04 u~~OPP Pr,c:tt|Sr!)hashreprrLs r*__hash__zEcuState.__hash__sDJJr,czd|_t|j}|D] }|j|= dSr!)rr$rr5)r&r5r(s r*resetzEcuState.resetsJDM&&(()) ! !A a   ! !r,cdddt|jdDzdzS)Nz EcuState(, c<g|]\}}|dt|S)=)r~rWs r*r3z$EcuState.command..s9 < < <14777 # < < .s QqTr,r^)r_rLs r*commandzEcuState.commandshTYY < < ##%%>>1;1;1; < < <===?BC Cr,ctjdkrtdtdrtdfd}|S)z Decorator to add a function as 'modify_ecu_state' method to a given class. This allows dynamic modifications and additions to a protocol. :param cls: A packet class to be modified :return: Decorator function rz)Packets without fields can't be extended.modify_ecu_statez7Class already extended. Can't override existing method.c*td|dS)Nrsetattrfclss r*decorator_functionz=EcuState.extend_pkt_with_modifier..decorator_functions C+Q / / / / /r,)rD fields_descr rCrrs` r*extend_pkt_with_modifierz!EcuState.extend_pkt_with_modifiersv s  1 $ $!"MNN N 3* + + K!IKK K 0 0 0 0 0"!r,cXtd|DS)z Helper function to determine if a Packet contains a layer that modifies the EcuState. :param pkt: Packet to be analyzed :return: True if pkt contains layer that implements modify_ecu_state c36K|]}t|dVdS)rN)rC)r2layers r*rXz+EcuState.is_modifier_pkt..sA..5"455......r,)anylayers)pkts r*is_modifier_pktzEcuState.is_modifier_pkts:.. # ..... .r,Fc|r|}ntj|}|D]W}t|ds ||||,#t$r|j|||YTwxYw|S)ax Helper function to get a modified EcuState from a Packet and a previous EcuState. An EcuState is always modified after a response Packet is received. In some protocols, the belonging request packet is necessary to determine the precise state of the Ecu :param response: Response packet that supports `modify_ecu_state` :param request: Belonging request of the response that modifies Ecu :param state: The previous/current EcuState :param modify_in_place: If True, the given EcuState will be modified :return: The modified EcuState or a modified copy r)copyrrCrrtim_func)responserequeststatemodify_in_place new_staters r*get_modified_ecu_statezEcuState.get_modified_ecu_states  )II %((I__&& M ME5"455  M&&x)DDDD M M M&..x)LLLLL MsA&BBN)F)__name__ __module__ __qualname____doc__ __slots__r+r= staticmethodr1rJr@rQr%rbrirmror{rrrrrrr0r,r*rr-sz [)I###!!! \  )))###### JJJ   @@@ !!!%P%P%PN   !!!CCC ""\"* . .\ .\r,rceZdZdZ ddZdZdZdZdZd Z e d Z e d Z e d Zd Ze dZdS)ra^An Ecu object can be used to * track the states of an Ecu. * to log all modification to an Ecu. * to extract supported responses of a real Ecu. Example: >>> print("This ecu logs, tracks and creates supported responses") >>> my_virtual_ecu = Ecu() >>> my_virtual_ecu.update(PacketList([...])) >>> my_virtual_ecu.supported_responses >>> print("Another ecu just tracks") >>> my_tracking_ecu = Ecu(logging=False, store_supported_responses=False) >>> my_tracking_ecu.update(PacketList([...])) >>> print("Another ecu just logs all modifications to it") >>> my_logging_ecu = Ecu(verbose=False, store_supported_responses=False) >>> my_logging_ecu.update(PacketList([...])) >>> my_logging_ecu.log >>> print("Another ecu just creates supported responses") >>> my_response_ecu = Ecu(verbose=False, logging=False) >>> my_response_ecu.update(PacketList([...])) >>> my_response_ecu.supported_responses Parameters to initialize an Ecu object :param logging: Turn logging on or off. Default is on. :param verbose: Turn tracking on or off. Default is on. :param store_supported_responses: Create a list of supported responses if True. :param lookahead: Configuration for lookahead when computing supported responses T ct|_||_||_||_||_t t|_t|_ t|_ dSr!) rrverboseloggingstore_supported_responses lookaheadrr$log_Ecu__supported_responsesr _Ecu__unanswered_packets)r&rrrrs r*r+z Ecu.__init__ sYZZ   )B&"t$$%)VV"$.LL!!!r,c0td|_dS)zB Resets the internal state to a default EcuState. r.sessionN)rrrLs r*rz Ecu.reset,s a((( r,ct|tr|D]}||dSt|tst d||dS)z Processes a Packet or a list of Packets, according to the chosen configuration. :param p: Packet or list of Packets z%Provide a Packet object for an updateN)r#r updaterrt _Ecu__update)r&prs r*rz Ecu.update3s} a $ $  ! ! C     ! !Av&& CDD D MM!     r,c|jr*tt|t||jr||||dS)zv Processes a Packet according to the chosen configuration. :param pkt: Packet to be processed N)rprintr~r_Ecu__update_log _Ecu__update_supported_responsesr&rs r*__updatez Ecu.__updateBsb < ) $t**d3ii ( ( ( < #   c " " " ))#.....r,c2|D]}t|ds ||\}}n-#t$r |j|\}}YnwxYw|j||j|fdS)aD Checks if a packet or a layer of this packet supports the function `get_log`. If `get_log` is supported, this function will be executed and the returned log information is stored in the intern log of this Ecu object. :param pkt: A Packet to be processed for log information. get_logN)rrCrrtrrr9time)r&rrlog_key log_values r* __update_logzEcu.__update_logNsZZ\\ < /2B2BC ,//$./II*$.!! 8 8ICE Idj11M  + +D#tz4 H H H1  6  80002 Jho== ..}=== EE 1 "=DAAAH| .fd8nn---  & - -h 7 7 7 7- 8 8r,ctt|jd}tt |d}|dk|dt |jpgz dt |jz fS)a. This sorts responses in the following order: 1. Positive responses first 2. Lower ServiceIDs first 3. Less supported states first 4. Longer (more specific) responses first :param resp: EcuResponse to be sorted :return: Tuple as sort key rl)rrrrrBrDr)r first_layerservices r* sort_key_funczEcu.sort_key_funcss64#4Q#788 eK((+,,4S!2333S!23335 5r,cP|j|j|jS)a< Returns a sorted list of supported responses. 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 r^)rsortrrLs r*supported_responseszEcu.supported_responsess* "''D,>'???))r,c|jS)z A list of all unanswered packets, which were processed by this Ecu object. :return: PacketList of unanswered packets )rrLs r*unanswered_packetszEcu.unanswered_packetss ((r,c*t|jSr!)r~rrLs r*rbz Ecu.__repr__sDJr,cfd}|S)z Decorator to add a function as 'get_log' method to a given class. This allows dynamic modifications and additions to a protocol. :param cls: A packet class to be modified :return: Decorator function c*td|dS)Nrrrs r*rz7Ecu.extend_pkt_with_logging..decorator_functions CA & & & & &r,r0rs` r*extend_pkt_with_loggingzEcu.extend_pkt_with_loggings$ ' ' ' ' '"!r,N)TTTr)rrrrr+rrrrrrrpropertyrrrbrr0r,r*rrs:.2;= 1 1 1 1)))    / / /<<<&%8%8%8N55\5$ * *X *))X)    " "\ " " "r,rc>eZdZdZfdZdedeefdZxZS)ra Tracks modification to an Ecu object 'on-the-flow'. The parameters for the internal Ecu object are obtained from the kwargs dict. `logging`: Turn logging on or off. Default is on. `verbose`: Turn tracking on or off. Default is on. `store_supported_responses`: Create a list of supported responses, if True. Example: >>> sniff(session=EcuSession) ct|dd|dd|dd|_tt|j|i|dS)NrTrr)rrr)rpopecusuperrr+)r&argsr' __class__s r*r+zEcuSession.__init__szvzz)T::%zz)T::17>> EcuResponse(EcuState(session=2, security_level=2), responses=UDS()/UDS_RDBIPR(dataIdentifier=2)/Raw(b"deadbeef1")) >>> EcuResponse([EcuState(session=range(2, 5), security_level=2), EcuState(session=3, security_level=5)], responses=UDS()/UDS_RDBIPR(dataIdentifier=9)/Raw(b"deadbeef4")) Initialize an EcuResponse capsule :param state: EcuState or list of EcuStates in which this response is allowed to be sent. If no state provided, the response packet will always be send. :param responses: A Packet or a list of Packet objects. By default the last packet is asked if it answers an incoming packet. This allows to send for example `requestCorrectlyReceived-ResponsePending` packets. :param answers: Optional argument to provide a custom answer here: `lambda resp, req: return resp.answers(req)` This allows the modification of a response depending on a request. Custom SecurityAccess mechanisms can be implemented in this way or generic NegativeResponse messages which answers to everything can be realized in this way. Nsc|d|_n@t|dr(ttt|}||_n|g|_t |t r||_nt |trt |g|_ndt|dr5ttt|}t ||_ntdt|z||_ dS)Nr?z Can't handle type %s as response) _EcuResponse__statesrCrrrr#r _EcuResponse__responsesrrtrs_EcuResponse__custom_answers)r&rranswerss r*r+zEcuResponse.__init__s = DMMuj)) (T(^U33 % !& i , , F(D    6 * * F)9+66D   Y + + FT&\955I))44D  2T)__DFF F!(r,c|jSr!)rrLs r*rzEcuResponse.statess }r,c|jSr!rrLs r*rzEcuResponse.responsess r,c |jd}|S)Nrrs r*rzEcuResponse.key_response sr" r,c|jt|jdkrdStfd|jDS)NrTc3,K|]}|kp|vVdSr!r0)r2rlrs r*rXz-EcuResponse.supports_state..+s0GGAqEz/UaZGGGGGGr,)rrDr)r&rs `r*supports_statezEcuResponse.supports_state&sI = C $6$6!$;$;4GGGGGGGGG Gr,cz|j||j|S|j|Sr!)rrrrhs r*rzEcuResponse.answers-s;  ,(():EBB B$,,U33 3r,cRt|jdd|jDS)N , responses=c6g|]}|Sr0)summary)r2rs r*r3z(EcuResponse.__repr__..8s ===D===r,)r~rrrLs r*rbzEcuResponse.__repr__4s8T]####==D,<====? ?r,c:ttt|jtjko0t dt |jjD}|j|Stfd|jDo|S)Nc3\K|]'\}}t|t|kV(dSr!)rB)r2r;rFs r*rXz%EcuResponse.__eq__..@sOGGAaE!HH$GGGGGGr,c3BK|]}|VdSr!)r)r2rlres r*rXz%EcuResponse.__eq__..Es1FF1u++A..FFFFFFr,)rrrDrrfrvrr)r&reresponses_equals ` r*rizEcuResponse.__eq__:s[%((   3u#7#7 7 G GGT^5:_2F2FGGG G G  = " "FFFF FFFFF  r,c||k Sr!r0rhs r*rozEcuResponse.__ne__Hs5=  r,c|jYdddd|jDzdzdddd|jDzdzdSd dzdd |jDzdzS) Nz EcuResponse([rc3>K|]}|VdSr!r)r2rls r*rXz&EcuResponse.command..Qs*CC CCCCCCr,]rc3>K|]}|VdSr!r r2rs r*rXz&EcuResponse.command..Rs*FF FFFFFFr,rzEcuResponse(responses=%s)c3>K|]}|VdSr!r rs r*rXz&EcuResponse.command..TsCA7A7 ! A7A7A7A7A7A7r,)rr`rrLs r*rzEcuResponse.commandMs = $ $diiCCT]CCCCCCcIIIdiiFFT5EFFFFFFLLLN N/4tyyA7A7%)%5A7A7A7878779<= =r,)rrrrrr+rrrrrrrbrirorrr0r,r*rrs2"SS-=-=t((((0X  X X HHH444???    !!! ===HHHr,rcbeZdZdZdZgdZdddeddfdZedZ dZ dZ dZ d d Z dS) raAnsweringMachine which emulates the basic behaviour of a real world ECU. Provide a list of ``EcuResponse`` objects to configure the behaviour of a AnsweringMachine. Usage: >>> resp = EcuResponse(session=range(0,255), security_level=0, responses=UDS() / UDS_NR(negativeResponseCode=0x7f, requestServiceId=0x10)) >>> sock = ISOTPSocket(can_iface, tx_id=0x700, rx_id=0x600, basecls=UDS) >>> answering_machine = EcuAnsweringMachine(supported_responses=[resp], main_socket=sock, basecls=UDS) >>> sim = threading.Thread(target=answering_machine, kwargs={'count': 4, 'timeout':5}) >>> sim.start() )store opened_socketcountfilterprn stop_filtertimeoutNcD||_|jg|_||j||ptd|_t |_tj|j|_||_ ||_ ||j d<|j|j d<dS)a :param supported_responses: List of ``EcuResponse`` objects to define the behaviour. The default response is ``generalReject``. :param main_socket: Defines the object of the socket to send and receive packets. :param broadcast_socket: Defines the object of the broadcast socket. Listen-only, responds with the main_socket. `None` to disable broadcast capabilities. :param basecls: Provide a basecls of the used protocol :param timeout: Specifies the timeout for sniffing in seconds. Nr.rrr) _main_socket_socketsr9r_initial_ecu_stater_ecu_state_mutexr _ecu_state_basecls_supported_responses sniff_options)r&r main_socketbroadcast_socketbaseclsrinitial_ecu_states r* parse_optionsz!EcuAnsweringMachine.parse_optionsjs,(*+  ' M !1 2 2 2"3"Jx7J7J7J $)D$;<< $7!(/9%.2m?+++r,c|jSr!)rrLs r*rzEcuAnsweringMachine.states r,c|j5tj|j|_ddddS#1swxYwYdSr!)rrrrrLs r* reset_statezEcuAnsweringMachine.reset_states  " A A"i(?@@DO A A A A A A A A A A A A A A A A A As 488c,t||jSr!)r#r)r&rs r* is_requestzEcuAnsweringMachine.is_requests#t}---r,c|j|jD]}t|tstd|j5||js dddU||s dddwt |j ||jd|j cdddcS#1swxYwYt| dt|ddzdzgS)a Checks if a given request can be answered by the internal list of EcuResponses. First, it's evaluated if the internal EcuState of this AnsweringMachine is supported by an EcuResponse, next it's evaluated if a request answers the key_response of this EcuResponse object. The first fitting EcuResponse is used. If this EcuResponse modified the EcuState, the internal EcuState of this AnsweringMachine is updated, and the list of response Packets of the selected EcuResponse is returned. If no EcuResponse if found, a PacketList with a generic NegativeResponse is returned. :param req: A request packet :return: A list of response packets Nz@Unsupported type for response. Please use `EcuResponse` objects.Trr.)r r#rrtrrrrrrrrr rrB)r&rrs r* make_replyzEcuAnsweringMachine.make_replys  $ 01 * *!$ 44I#%HIII* * *..t??!  * * * * * * * <<,,! * * * * * * *33)3GGG > * * * * * * * * * * * * * * * * * * *4== eCjj1o % /11233 3sC$C-CC C c|D]]}t|dkr'tjtjdd|jr|j|^dS)a7 Sends all Packets of a EcuResponse object. This allows to send multiple packets up on a request. If the list contains more than one packet, a random time between each packet is waited until the next packet will be sent. :param reply: List of packets to be sent. r.g{Gz?g?N)rDrsleeprandomuniformrsend)r&reply send_functionrs r* send_replyzEcuAnsweringMachine.send_replysk * *A5zzA~~ 6>$44555  *!&&q)))  * *r,r!)rrrr function_namesniff_options_listrr&rrr)r+r/r7r0r,r*rrZs  *M444 !%!"$<$<$<$rGsv ######$$$$$$$$""""""))))))------))))))''''''                         " " "RRRRRvRRRjB"B"B"B"B"&B"B"B"J:sssssssslu*u*u*u*u**:6u*u*u*u*u*r,