(pcap, any_port=False)
return results
Process packets within the PCAP.

    :param pcap: Opened PCAP file to be processed
    :type pcap: dpkt.pcap.Reader
    :param any_port: Whether or not to search for non-SSL ports
    :type any_port: bool
Process packets within the PCAP.
Process packets within the PCAP .
def process_pcap(pcap, any_port=False):
    """Process packets within the PCAP.

    :param pcap: Opened PCAP file to be processed
    :type pcap: dpkt.pcap.Reader
    :param any_port: Whether or not to search for non-SSL ports
    :type any_port: bool
    """
    decoder = dpkt.ethernet.Ethernet
    linktype = pcap.datalink()
    if linktype == dpkt.pcap.DLT_LINUX_SLL:
        decoder = dpkt.sll.SLL
    elif linktype == dpkt.pcap.DLT_NULL or linktype == dpkt.pcap.DLT_LOOP:
        decoder = dpkt.loopback.Loopback

    results = list()
    for timestamp, buf in pcap:
        try:
            eth = decoder(buf)
        except Exception:
            continue

        if not isinstance(eth.data, (dpkt.ip.IP, dpkt.ip6.IP6)):
            # We want an IP packet
            continue
        if not isinstance(eth.data.data, dpkt.tcp.TCP):
            # TCP only
            continue

        ip = eth.data
        tcp = ip.data

        if not (tcp.dport == SSL_PORT or tcp.sport == SSL_PORT or any_port):
            # Doesn't match SSL port or we are picky
            continue
        if len(tcp.data) <= 0:
            continue

        tls_handshake = bytearray(tcp.data)
        if tls_handshake[0] != TLS_HANDSHAKE:
            continue

        records = list()

        try:
            records, bytes_used = dpkt.ssl.tls_multi_factory(tcp.data)
        except dpkt.ssl.SSL3Exception:
            continue
        except dpkt.dpkt.NeedData:
            continue

        if len(records) <= 0:
            continue

        for record in records:
            if record.type != TLS_HANDSHAKE:
                continue
            if len(record.data) == 0:
                continue
            server_hello = bytearray(record.data)
            if server_hello[0] != 2:
                # We only want server HELLO
                continue
            try:
                handshake = dpkt.ssl.TLSHandshake(record.data)
            except dpkt.dpkt.NeedData:
                # Looking for a handshake here
                continue
            if not isinstance(handshake.data, dpkt.ssl.TLSServerHello):
                # Still not the HELLO
                continue

            server_handshake = handshake.data
            ja3 = [str(server_handshake.version)]

            # Cipher Suites (16 bit values)
            ja3.append(str(server_handshake.cipher_suite))
            ja3 += process_extensions(server_handshake)
            ja3 = ",".join(ja3)

            record = {"source_ip": convert_ip(ip.src),
                      "destination_ip": convert_ip(ip.dst),
                      "source_port": tcp.sport,
                      "destination_port": tcp.dport,
                      "ja3": ja3,
                      "ja3_digest": md5(ja3.encode()).hexdigest(),
                      "timestamp": timestamp}
            results.append(record)

    return results
def process_pcap ( pcap , any_port = False ) : decoder = dpkt . ethernet . Ethernet linktype = pcap . datalink ( ) if linktype == dpkt . pcap . DLT_LINUX_SLL : decoder = dpkt . sll . SLL elif linktype == dpkt . pcap . DLT_NULL or linktype == dpkt . pcap . DLT_LOOP : decoder = dpkt . loopback . Loopback results = list ( ) for timestamp , buf in pcap : try : eth = decoder ( buf ) except Exception : continue if not isinstance ( eth . data , ( dpkt . ip . IP , dpkt . ip6 . IP6 ) ) : # We want an IP packet continue if not isinstance ( eth . data . data , dpkt . tcp . TCP ) : # TCP only continue ip = eth . data tcp = ip . data if not ( tcp . dport == SSL_PORT or tcp . sport == SSL_PORT or any_port ) : # Doesn't match SSL port or we are picky continue if len ( tcp . data ) <= 0 : continue tls_handshake = bytearray ( tcp . data ) if tls_handshake [ 0 ] != TLS_HANDSHAKE : continue records = list ( ) try : records , bytes_used = dpkt . ssl . tls_multi_factory ( tcp . data ) except dpkt . ssl . SSL3Exception : continue except dpkt . dpkt . NeedData : continue if len ( records ) <= 0 : continue for record in records : if record . type != TLS_HANDSHAKE : continue if len ( record . data ) == 0 : continue server_hello = bytearray ( record . data ) if server_hello [ 0 ] != 2 : # We only want server HELLO continue try : handshake = dpkt . ssl . TLSHandshake ( record . data ) except dpkt . dpkt . NeedData : # Looking for a handshake here continue if not isinstance ( handshake . data , dpkt . ssl . TLSServerHello ) : # Still not the HELLO continue server_handshake = handshake . data ja3 = [ str ( server_handshake . version ) ] # Cipher Suites (16 bit values) ja3 . append ( str ( server_handshake . cipher_suite ) ) ja3 += process_extensions ( server_handshake ) ja3 = "," . join ( ja3 ) record = { "source_ip" : convert_ip ( ip . src ) , "destination_ip" : convert_ip ( ip . dst ) , "source_port" : tcp . sport , "destination_port" : tcp . dport , "ja3" : ja3 , "ja3_digest" : md5 ( ja3 . encode ( ) ) . hexdigest ( ) , "timestamp" : timestamp } results . append ( record ) return results