a
    kÄ#dÙO  ã                   @   s@  d dl Z d dlZd dlmZ d dlmZmZmZmZm	Z	m
Z
 d dlmZ d dlmZ d dlmZ d dlmZ d dlmZ d d	lmZmZmZmZ e  d
¡Zdd„ Zd'dd„Zdd„ Zdd„ Zedƒdddddfdd„Z edƒdedƒddddfdd„Z!edƒdedƒddddddddfdd„Z"d(d d!„Z#d)d#d$„Z$d*d%d&„Z%dS )+é    N)ÚEvent)ÚIterableÚOptionalÚUnionÚListÚTupleÚDict)ÚPacket)Úorb)ÚCAN)ÚSuperSocket)Ú
PYTHON_CAN)ÚISOTPHeaderÚISOTPHeaderEAÚISOTP_FFÚISOTPzscapy.contrib.isotpc                 C   s6   t || dƒ}t||d ƒD ]}||_|  |¡ qdS )a+  Send multiple packets with extended addresses at once.

    This function is used for scanning with extended addresses.
    It sends multiple packets at once. The number of packets
    is defined in the number_of_packets variable.
    It only iterates the extended ID, NOT the actual CAN ID of the packet.
    This method is used in extended scan function.

    :param sock: CAN interface to send packets
    :param ext_id: Extended ISOTP-Address
    :param packet: Template Packet
    :param number_of_packets: number of packets to send in one batch
    éÿ   é   N)ÚminÚrangeÚextended_addressÚsend)ÚsockÚext_idÚpacketZnumber_of_packetsZend_idÚi© r   úX/var/www/html/gsl.tools/lib/python3.9/site-packages/scapy/contrib/isotp/isotp_scanner.pyÚsend_multiple_ext   s    r   Fc                 C   sJ   |rt ƒ tƒ  }d|_d|_ntƒ tƒ  }d|_|r:d|_| |_d|_|S )zéCraft ISO-TP packet

    :param identifier: identifier of crafted packet
    :param extended: boolean if packet uses extended address
    :param extended_can_id: boolean if CAN should use extended Ids
    :return: Crafted Packet
    r   s        s         Úextendedéd   )r   r   r   Údatar   ÚflagsÚ
identifierZmessage_size)r#   r   Úextended_can_idÚpktr   r   r   Úget_isotp_packet0   s    
r&   c           
      C   sþ   i }|   ¡ D ]V\}}|d }|d }||vr>|g|gf||< q|| \}}||g ||g f||< q|D ]�}|| d }|| d }t|ƒdk r’qhdd„ t|dd… |dd… ƒD ƒ}tdd	„ t|dd… |dd… ƒD ƒƒrht d
¡ |D ]
}	| |	= qìqhdS )zìFilter to remove periodic packets from packet_dict

    ISOTP-Filter for periodic packets (same ID, always same time-gaps)
    Deletes periodic packets in packet_dict

    :param packet_dict: Dictionary, where the filter is applied
    r   r   é   c                 S   s$   g | ]\}}t |jƒt |jƒ ‘qS r   )ÚfloatÚtime)Ú.0Úp1Úp2r   r   r   Ú
<listcomp>c   s   ÿz+filter_periodic_packets.<locals>.<listcomp>Néÿÿÿÿc                 s   s"   | ]\}}t || ƒd k V  qdS )gü©ñÒMbP?N)Úabs)r*   Út1Út2r   r   r   Ú	<genexpr>e   ó    z*filter_periodic_packets.<locals>.<genexpr>z5[i] Identifier 0x%03x seems to be periodic. Filtered.)ÚitemsÚlenÚzipÚallÚ	log_isotpÚinfo)
Zpacket_dictZfilter_dictÚkeyÚvaluer%   ZidnZkey_lstZpkt_lstÚtgÚkr   r   r   Úfilter_periodic_packetsI   s,    	ÿ(ÿr>   c           	   
   C   s  |j r|j dkrdS |dur*|j|v r*dS z²|r4dnd}t|j| ƒd? }t|j| ƒd@ }|dkrÆd|  krxdkrÆn nJt d	|j| ¡ t|tƒr¦||jf|| < qÚt|tƒr¼| 	| ¡ qÚt
d
ƒ‚n|durÚ| 	|j¡ W n8 t�y } zt d|t|ƒ¡ W Y d}~n
d}~0 0 dS )a¬  Callback for sniff function when packet received

    If received packet is a FlowControl and not in noise_ids append it
    to id_list.

    :param id_value: packet id of send packet
    :param id_list: list of received IDs
    :param noise_ids: list of packet IDs which will not be considered when
                      received during scan
    :param extended: boolean if extended scan
    :param packet: received packet
    r   Nr   r   é   é   r'   é   zNFound flow-control frame from identifier 0x%03x when testing identifier 0x%03xzUnknown type of id_listz+Unknown message Exception: %s on packet: %s)r"   r#   r
   r!   r8   r9   Ú
isinstanceÚdictÚlistÚappendÚ	TypeErrorÚ	ExceptionÚ	exceptionÚrepr)	Zid_valueÚid_listÚ	noise_idsr   r   ÚindexZ	isotp_pciZisotp_fcÚer   r   r   Úget_isotp_fcl   s0     þ


þrN   i   gš™™™™™¹?Tc                    sú   t ƒ ‰|D ]T‰|dur"| ¡ r" q`ˆr0ˆˆv r0q
|  tˆd|ƒ¡ | j‡‡‡fdd„|dd� q
|shˆS t ƒ ‰ ˆ ¡ D ]~}|durŽ| ¡ rŽ qöttd|d ƒ|d dƒD ]J‰|durÀ| ¡ rÀ qv|  tˆd|ƒ¡ | j‡ ‡‡fd	d„|d
 dd� q¨qvˆ S )a6  Scan and return dictionary of detections

    ISOTP-Scan - NO extended IDs
    found_packets = Dictionary with Send-to-ID as
    key and a tuple (received packet, Recv_ID)

    :param sock: socket for can interface
    :param scan_range: hexadecimal range of IDs to scan. Default is 0x0 - 0x7ff
    :param noise_ids: list of packet IDs which will not be tested during scan
    :param sniff_time: time the scan waits for isotp flow control responses
                       after sending a first frame
    :param extended_can_id: Send extended can frames
    :param verify_results: Verify scan results. This will cause a second scan
                           of all possible candidates for ISOTP Sockets
    :param stop_event: Event object to asynchronously stop the scan
    :return: Dictionary with all found packets
    NFc                    s   t ˆˆˆ d| ƒS ©NF©rN   ©r%   )rK   Úreturn_valuesr;   r   r   Ú<lambda>½   s   ÿzscan.<locals>.<lambda>©ZprnÚtimeoutÚstorer   rA   r   c                    s   t ˆˆ ˆd| ƒS rO   rP   rQ   )Úcleaned_ret_valrK   r;   r   r   rS   Ì   s   ÿé
   )rC   Úis_setr   r&   ÚsniffÚkeysr   Úmax)r   Ú
scan_rangerK   Ú
sniff_timer$   Úverify_resultsÚ
stop_eventZ	tested_idr   )rW   rK   rR   r;   r   Úscan�   s0    þþ
ra   é    é   c                    sR  t ƒ ‰|pd}|D �]8}ˆr&|ˆv r&qt|d|d�}	g ‰t|ƒ}
t|
d |
d |ƒD ]P‰ |durl| ¡ rl q¦t| ˆ |	|ƒ | j‡ ‡‡fdd„|d	 d
d� t |¡ qTtt	ˆƒƒ‰ˆD ]”‰ |durÎ| ¡ rÎ qtt
ˆ d dƒtˆ | d dƒƒD ]X}|du�r| ¡ �r q¶||	_|d> | ‰|  |	¡ | j‡‡‡fdd„|d d
d� qðq¶qˆS )a�  Scan with ISOTP extended addresses and return dictionary of detections

    If an answer-packet found -> slow scan with
    single packages with extended ID 0 - 255
    found_packets = Dictionary with Send-to-ID
    as key and a tuple (received packet, Recv_ID)

    :param sock: socket for can interface
    :param scan_range: hexadecimal range of IDs to scan. Default is 0x0 - 0x7ff
    :param scan_block_size: count of packets send at once
    :param extended_scan_range: range to search for extended ISOTP addresses
    :param noise_ids: list of packet IDs which will not be tested during scan
    :param sniff_time: time the scan waits for isotp flow control responses
                       after sending a first frame
    :param extended_can_id: Send extended can frames
    :param stop_event: Event object to asynchronously stop the scan
    :return: Dictionary with all found packets
    r   T)r   r$   r   r.   Nc                    s   t ˆ ˆˆd| ƒS ©NTrP   )Úp)Úext_isotp_idrJ   rK   r   r   rS   ý   s   ÿzscan_extended.<locals>.<lambda>r'   FrT   rA   rc   é   c                    s   t ˆ ˆˆd| ƒS rd   rP   rQ   )Úfull_idrK   rR   r   r   rS     s   ý)rC   r&   rD   r   rY   r   rZ   r)   ÚsleepÚsetr\   r   r   r   )r   r]   Zscan_block_sizeÚextended_scan_rangerK   r^   r$   r`   r;   r%   Úrr   r   )rf   rh   rJ   rK   rR   r   Úscan_extendedÓ   sD    
ÿþÿ
ürm   rA   c              	      sÄ   |
rt  tj¡ t  d¡ tddd�‰ ˆj|‡ ‡fdd„d�}ttdd	„ |D ƒƒƒ}|rpt	ˆ||||||d
�}nt
ˆ|||||	|d�}t|ƒ |dkr t||ƒS |dkr´t|||ƒS t||p¾ˆ|ƒS )aä  Scan for ISOTP Sockets on a bus and return findings

    Scan for ISOTP Sockets in the defined range and returns found sockets
    in a specified format. The format can be:

    - text: human readable output
    - code: python code for copy&paste
    - sockets: if output format is not specified, ISOTPSockets will be
      created and returned in a list

    :param sock: CANSocket object to communicate with the bus under scan
    :param scan_range: range of CAN-Identifiers to scan. Default is 0x0 - 0x7ff
    :param extended_addressing: scan with ISOTP extended addressing
    :param extended_scan_range: range for ISOTP extended addressing values
    :param noise_listen_time: seconds to listen for default communication on
                              the bus
    :param sniff_time: time the scan waits for isotp flow control responses
                       after sending a first frame
    :param output_format: defines the format of the returned results
                          (text, code or sockets). Provide a string e.g.
                          "text". Default is "socket".
    :param can_interface: interface used to create the returned code/sockets
    :param extended_can_id: Use Extended CAN-Frames
    :param verify_results: Verify scan results. This will cause a second scan
                           of all possible candidates for ISOTP Sockets
    :param verbose: displays information during scan
    :param stop_event: Event object to asynchronously stop the scan
    :return:
    zFiltering background noise...i#  s   ª»ÌÝîÿª»)r#   r!   c                      s
   ˆ  ˆ ¡S ©N)r   r   ©Z	dummy_pktr   r   r   rS   O  r3   zisotp_scan.<locals>.<lambda>)rU   Zstarted_callbackc                 s   s   | ]}|j V  qd S rn   )r#   )r*   r%   r   r   r   r2   Q  r3   zisotp_scan.<locals>.<genexpr>)rk   rK   r^   r$   r`   )rK   r^   r$   r_   r`   ÚtextÚcode)r8   ÚsetLevelÚloggingÚDEBUGr9   r   rZ   rD   rj   rm   ra   r>   Úgenerate_text_outputÚgenerate_code_outputÚgenerate_isotp_list)r   r]   Úextended_addressingrk   Znoise_listen_timer^   Zoutput_formatÚcan_interfacer$   r_   Úverboser`   Zbackground_pktsrK   Úfound_packetsr   ro   r   Ú
isotp_scan  sH    +
ÿþûû
ÿ
ÿr|   c              
   C   sî   | sdS dt | ƒ }| D ]Ð}|rˆ|d }||d  }tt| | d jd ƒƒ}|dt|ƒt|ƒt| | d jƒ|t| | d ƒf 7 }n0|dt|ƒt| | d jƒt| | d ƒf 7 }| | d jdk}|rØ|d7 }n|d	7 }|d
7 }q|S )aV  Generate a human readable output from the result of the `scan` or the
    `scan_extended` function.

    :param found_packets: result of the `scan` or `scan_extended` function
    :param extended_addressing: print results from a scan with
                                ISOTP extended addressing
    :return: human readable scan results
    zNo packets found.z&
Found %s ISOTP-FlowControl Packet(s):rc   r   z‡
Send to ID:             %s
Send to extended ID:    %s
Received ID:            %s
Received extended ID:   %s
Message:                %szQ
Send to ID:             %s
Received ID:            %s
Message:                %srg   z
Padding enabledz
No PaddingÚ
)r5   Úhexr
   r!   r#   rI   Úlength)r{   rx   rp   ÚpackÚsend_idÚsend_extr   Úpaddingr   r   r   ru   o  s4    
þû	þý

ru   Úifacec           	   	   C   sº   d}| s|S dt  }| D ]˜}|r€|d }||d  }t| | d jd ƒ}|d||t| | d jƒ| | d jdk||f 7 }q|d||t| | d jƒ| | d jdkf 7 }q|| S )aô  Generate a copy&past-able output from the result of the `scan` or
    the `scan_extended` function.

    :param found_packets: result of the `scan` or `scan_extended` function
    :param can_interface: description string for a CAN interface to be
                          used for the creation of the output.
    :param extended_addressing: print results from a scan with ISOTP
                                extended addressing
    :return: Python-code as string to generate all found sockets
    Ú zr

import can
conf.contribs['CANSocket'] = {'use-python-can': %s}
load_contrib('cansocket')
load_contrib('isotp')

rc   r   zjISOTPSocket(%s, tx_id=0x%x, rx_id=0x%x, padding=%s, ext_address=0x%x, rx_ext_address=0x%x, basecls=ISOTP)
rg   zCISOTPSocket(%s, tx_id=0x%x, rx_id=0x%x, padding=%s, basecls=ISOTP)
)r   r
   r!   Úintr#   r   )	r{   ry   rx   ÚresultÚheaderr€   r�   r‚   r   r   r   r   rv   œ  s4    ýüý
þþrv   c                 C   s¦   ddl m} g }| D ]Œ}| | d }|j}|jdkr8dnd}|r„|d? }	t||	d  ƒ}
t|jd ƒ}| |||	|
|||td�¡ q|}	| |||	||td�¡ q|S )	aç  Generate a list of ISOTPSocket objects from the result of the `scan` or
    the `scan_extended` function.

    :param found_packets: result of the `scan` or `scan_extended` function
    :param can_interface: description string for a CAN interface to be
                          used for the creation of the output.
    :param extended_addressing: print results from a scan with ISOTP
                                extended addressing
    :return: A list of all found ISOTPSockets
    r   )ÚISOTPSocketrg   TFrc   )Útx_idZext_addressÚrx_idZrx_ext_addressrƒ   Úbasecls)rŠ   r‹   rƒ   rŒ   )	Zscapy.contrib.isotpr‰   r#   r   r†   r
   r!   rE   r   )r{   ry   rx   r‰   Zsocket_listr€   r%   Zdest_idÚpadZ	source_idZ
source_extZdest_extr   r   r   rw   É  s.    
û

þ
rw   )r   FF)F)r„   F)F)&rs   r)   Ú	threadingr   Zscapy.compatr   r   r   r   r   r   Zscapy.packetr	   r
   Zscapy.layers.canr   Zscapy.supersocketr   Zscapy.contrib.cansocketr   Z scapy.contrib.isotp.isotp_packetr   r   r   r   Ú	getLoggerr8   r   r&   r>   rN   r   ra   rm   r|   ru   rv   rw   r   r   r   r   Ú<module>	   s^    

#2ú
7ù
Fõ
W
-  ÿ
/ þ