Using producer consumer model and dpkt to process pcap files

1、 Use find Command to generate pcap File list file

2、 Use dpkt extract pcap Source address source port destination address destination port , The printing program takes time

find route -type f -name *.pcap >pcaplist.txt

#!/bin/env python3
import time
from multiprocessing import Process,JoinableQueue,cpu_count
import dpkt
import datetime
from dpkt.utils import mac_to_str,inet_to_str
# send dpkt extract pcap Information about each in the document
def print_packets(pcap):
    for timestamp,buf in pcap:
        print('Timestamp: ',str(datetime.datetime.utcfromtimestamp(timestamp)))
        eth = dpkt.ethernet.Ethernet(buf)
        print('Ethernet Frame: ',mac_to_str(eth.src),mac_to_str(eth.dst),eth.type)
        if not isinstance(eth.data,dpkt.ip.IP):
            print('Non IP Packet type not supported %s\n'% eth.data.__class__.__name__)
            continue
        ip = eth.data
        print('IP: %s -> %s (len=%d ttl=%d DF=%d MF=%d offset=%d)\n' %(inet_to_str(ip.src),inet_to_str(ip.dst),ip.len,ip.ttl,ip.df,ip.mf,ip.offset))
    print('**Pretty print demo **\n')
    eth.pprint()
# Will read pcapfile file name , Write the file name to be processed to result_f
def analyze_pcap(pcapfile,result_f):
    pcapfile=pcapfile.strip()
    with open(pcapfile,'rb') as pcapfilef:
        pcap = dpkt.pcap.Reader(pcapfilef) # call dpkt, Read pcap file
        print_packets(pcap) # Function function , Print relevant information here , It can be replaced by other functions
    
    result_f.write(pcapfile) # After processing , To be dealt with pcap Write a file name , As log Record
    result_f.flush()# This must have , Save to hard disk
# A consumer function , Read the file list of the files to be processed
def consumer(queue,result_f):
    while True:
        pcapfile = queue.get()
        analyze_pcap(pcapfile,result_f)
        queue.task_done()
# A producer function , Read the file name from the list of files to be processed , Put in queue
def producer(queue):
    with open('pcapfilelist.txt','r') as f:
        for pcapfilename in f:
            queue.put(pcapfilename)

def main():
    with open('processed_pcapfile.txt','w+') as result_f:
        queue = JoinableQueue(40) # Limit the number of queues
        pc = Process(target=producer,args=(queue,))
        pc.start()

        for _ in range(cpu_count()):
            c1 = Process(target=consumer,args=(queue,result_f))
            c1.daemon = True
            c1.start()
        pc.join()
        queue.join()

if __name__ == '__main__':
    now = lambda:time.time()
    start = now()
    main()
    print("Time:",now() - start)