Python pcap分析时提取邮箱列表不更新、循环输出问题求助

• 邮箱存储逻辑错误：email_list函数每次被调用时都会初始化全新的空列表，仅存储当前单个报文匹配到的邮箱，无法累计存储所有报文的邮箱结果，且每次append后直接打印，才会出现你看到的逐行重复、零散的输出。
• 无去重逻辑：没有使用可自动去重的容器（如set）存储邮箱，重复出现的邮箱会被多次记录。
• 调用逻辑冗余：tabulate_table函数末尾额外调用了tcp()，相当于协议统计完成后又重新遍历了一遍pcap文件，重复执行邮箱提取逻辑，进一步加剧了重复输出的问题。
• 无最终汇总输出流程：所有邮箱匹配完成后，没有将汇总去重的结果统一打印为表格的逻辑。

核心修改点 #

1. 新增全局的邮箱去重存储容器，使用set类型自动去重
2. 改写email_list函数，将匹配到的邮箱存入全局set，不要在函数内部初始化存储容器、不要逐次打印
3. 删除tabulate_table末尾的tcp()冗余调用，统一pcap遍历流程，邮箱提取和协议统计可以在同一次pcap遍历中完成，避免重复读取文件
4. 所有报文遍历完成后，将去重后的邮箱转为列表，用tabulate打印规范表格

修改后可运行代码 #

import dpkt,socket,datetime,geoip2.database,re,sys,urllib.request,urllib
from tabulate import tabulate
from collections import Counter
from prettytable import PrettyTable

# 全局去重存储邮箱
all_emails = set()

def packet_type(pcap):
    other = []
    IP = []
    tcp = []
    udp = []
    igmp = []

    for ts, buf in pcap:
        eth = dpkt.ethernet.Ethernet(buf)
        ip = eth.data
        TCP = ip.data
        info = repr(TCP)
        # 同一次遍历中提取邮箱，避免重复读文件
        email_list(info)
        src = socket.inet_ntoa(ip.src)
        dst = socket.inet_ntoa(ip.dst)
        try:
            if eth.type != dpkt.ethernet.ETH_TYPE_IP:
                other.append(src)
            IP.append(ip.len)
            if ip.p == dpkt.ip.IP_PROTO_IGMP:
                igmp.append(ip.len)
            elif ip.p == dpkt.ip.IP_PROTO_TCP:
                tcp.append(ip.len)
            elif ip.p == dpkt.ip.IP_PROTO_UDP:
                udp.append(ip.len)
        except Exception as err:
            print(f'Oh no there has been an {err}')
            continue
    timestamp(tcp,udp,igmp)
    # 协议统计完成后打印邮箱表格
    print_email_table()


def timestamp(tcp,udp,igmp):
    tcp.sort()
    Tcp = len(tcp)
    TCP1st = tcp[0]
    TCP2nd = tcp[-1]
    TCPts = str(datetime.datetime.utcfromtimestamp(TCP1st))
    TCP2ts = str(datetime.datetime.utcfromtimestamp(TCP2nd))
    udp.sort()
    Udp = len(udp)
    UDP = udp[0]
    UDP2nd = udp[-1]
    UDPts = str(datetime.datetime.utcfromtimestamp(UDP))
    UDP2ts = str(datetime.datetime.utcfromtimestamp(UDP2nd))
    igmp.sort()
    Igmp = len(igmp)
    IGMP = igmp[0]
    IGMP2nd = igmp[-1]
    IGMPts = str(datetime.datetime.utcfromtimestamp(IGMP))
    IGMP2ts = str(datetime.datetime.utcfromtimestamp(IGMP2nd))
    mean_packet_length(tcp,udp,igmp,TCPts,TCP2ts,UDPts,UDP2ts,IGMPts,IGMP2ts,Tcp,Udp,Igmp)


def mean_packet_length(tcp,udp,igmp,TCPts,TCP2ts,UDPts,UDP2ts,IGMPts,IGMP2ts,Tcp,Udp,Igmp):
    tcpmean = sum(tcp) / len(tcp)
    tcp_mean = round(tcpmean)
    udpmean = sum(udp) / len(udp)
    udp_mean = round(udpmean)
    igmpmean = sum(igmp) / len(igmp)
    igmp_mean = round(igmpmean)
    tabulate_table(tcp_mean,udp_mean,igmp_mean,TCPts,TCP2ts,UDPts,UDP2ts,IGMPts,IGMP2ts,Tcp,Udp,Igmp)


def tabulate_table(tcp_mean,udp_mean,igmp_mean,TCPts,TCP2ts,UDPts,UDP2ts,IGMPts,IGMP2ts,Tcp,Udp,Igmp):
    table =[['TCP',Tcp,TCPts,TCP2ts,tcp_mean], ['UDP',Udp,UDPts, UDP2ts, udp_mean], ['IGMP',Igmp,IGMPts,IGMP2ts,igmp_mean]]
    headers = ['Protocol','Count', 'First_Timestamp', 'Last_Timestamp', 'Mean_Length']
    print(tabulate(table, headers, tablefmt='fancy_grid'))
    # 删掉冗余的tcp()调用


def email_list(info):
    emaillist = re.findall(r"[a-zA-Z0-9.]+@[a-zA-Z0-9.]+\.\w{2,4}", info)
    for em in emaillist:
        # 存入全局set自动去重
        all_emails.add(em)

def print_email_table():
    # 去重后的邮箱转为表格格式
    email_table = [[idx+1, email] for idx, email in enumerate(all_emails)]
    headers = ['序号', '邮箱地址']
    print("\n[+] 提取到的去重邮箱列表：")
    print(tabulate(email_table, headers, tablefmt='fancy_grid'))


def find_uri():
    found = False
    gif_uri = []
    pcapFile = r'C:\Users\snoopgrapes\Desktop\evidence-packet-analysis.pcap'
    with open(pcapFile, 'rb') as pcapfile:
        pcap = dpkt.pcap.Reader(pcapfile)
        for ts, buf in pcap:
            try:
                eth = dpkt.ethernet.Ethernet(buf)
                ip = eth.data
                tcp = ip.data
                http = dpkt.http.Request(tcp.data)
                if http.method == 'GET':
                    uri = http.uri.lower()
                    if '.gif' in uri:
                        gif_uri.append(uri)
                        found = True
            except Exception:
                pass
    print(f'Gif URI {gif_uri}')


def main():
    pcapFile = r'C:\Users\snoopgrapes\Desktop\evidence-packet-analysis.pcap'
    excludesrc = '146.176.164.91'
    f = open(pcapFile, 'rb')
    pcap = dpkt.pcap.Reader(f)
    reader = geoip2.database.Reader('C:\Program Files\Python39\Geo\Geo.mmdb')
    print(f'[*] analysing {pcapFile} for packets not source {excludesrc}')
    print('------------------------------------------------------------')
    packet_type(pcap)
    # 可选调用find_uri
    # find_uri()


if __name__ == '__main__':
    main()

补充优化点 #

• 原正则仅能匹配常规后缀的邮箱，若需要覆盖更多合法邮箱格式，可以调整正则规则
• 所有pcap读取逻辑统一使用main函数中传入的路径，避免硬编码多份路径后续修改遗漏

内容的提问来源于stack exchange，提问作者snoopstargrapes