批量提取防火墙日志中IP对间通信端口的Python实现需求

我有一份从防火墙导出的CSV日志文件，已经转换成包含10万条记录的列表嵌套格式。现在需要生成一个输出文件，展示任意两个IP之间的通信端口，格式示例如下：

1.1.1.1 到 2.2.2.2 端口(25, 53, 80)
2.2.2.2 到 1.1.1.1 端口(443, 123)

目前我已经能手动提取单个IP对的端口，但日志里有4个源IP、67个目的IP，总共268种IP组合，手动操作太繁琐，希望通过循环自动遍历所有IP对，收集对应端口并写入文件。

现有手动提取代码 #

#! python3
import csv
#Read the file and covert the CSV to a usable list format
csv_filename = 'data-for-csv-reader_no_parentheses_v3.csv'
#need to add operation to strip " from lines
#open file and read into a list of lists:
with open(csv_filename) as f:
    reader =csv.reader(f)
    lst = list(reader)


#Extract all Source IPs:
srcips =[]
for item in lst:
    srcips.append(item[0])

#Deduplicate source IPS:
srciplist = [*set(srcips)]
print("The number of source IPs is " + str(len(srciplist)) + ".")

#Strip 'srcip= off of entry (no longer needed, pherhaps)
srciplist_stripped = [j.strip('srcip=') for j in srciplist]
srciplist_stripped.sort()
print(srciplist_stripped)

#Extract all destination IPs:
dstips =[]
for item in lst:
    dstips.append(item[1])

#Deduplicate destination IPs:
dstiplist = [*set(dstips)]
print("The number of destination IPs is " + str(len(dstiplist)) + ".")

#Strip 'dstip= off of entry (no longer needed, pherhaps)
dstiplist_stripped = [j.strip('dstip=') for j in dstiplist]
dstiplist_stripped.sort()
print(dstiplist_stripped)

#Manual operation to get one source and one destination's ports:
port_list = []
for item in lst:
    if item[0] == srciplist_stripped[2] and item[1] == dstiplist_stripped[4]:
        port_list.append(item[3])

#Presents port list for the prior two IPs
port_list = [*set(port_list)]
print("Source IP:" + str(srciplist_stripped[2]) + " Destination IP:" + str(dstiplist_stripped[4]) + " Port_list :" + str(port_list))
print("The number of ports is " + str(len(port_list)) + ".")

现有代码运行输出 #

The number of source IPs is 4.
['1.1.1.1', '2.2.2.2', '3.3.3.3', '4.4.4.4']
The number of destination IPs is 67.
['7.7.7.7', '6.6.6.6', '5.5.5.5', <--omitted for brevity-->]
Source IP:1.1.1.1. Destination IP:2.2.2.2 Port_list :['dstport=644', 'dstport=1039',<--omitted for brevity-->]
The number of ports is 873.

初步尝试的代码雏形 #

#!python
import csv
from itertools import product
import os
import fileinput

#Specify the source and destination variables, the FW log file, final output file.
src, dst, = {}, set()
log_file = input("Enter Log File(be sure to include ".csv:")")
output_file = input("Enter the location and file name to send output to:")

#Temp file for data manipulation.
temp_output_file = "temp_output_file.txt"

#Create data structure
with open(log_file, "r") as f_in:
    reader = csv.reader(f_in)
    for row in reader:
        src.setdefault(row[0], {}).setdefault(row[1], {}).setdefault(row[3], {}).setdefault(row[4])
        dst.add(row[1])

#Print to screen if desired
#for s, d in product(src, dst):
    #print(f"Source IP: {s} Destination IP: {d} Port_list: {src[s].get(d, [])}")
    #print('\n') 

解决方案代码 #

#! python3
import csv
from itertools import product

def process_firewall_log(log_file, output_file):
    # 初始化数据结构：{源IP: {目的IP: {端口集合}}}
    ip_port_map = {}
    
    with open(log_file, 'r') as f:
        reader = csv.reader(f)
        for row in reader:
            # 提取并清理源IP、目的IP、端口（去除前缀和引号）
            src_ip = row[0].strip('srcip="')
            dst_ip = row[1].strip('dstip="')
            port = row[3].strip('dstport="')
            
            # 构建层级映射，自动去重端口
            if src_ip not in ip_port_map:
                ip_port_map[src_ip] = {}
            if dst_ip not in ip_port_map[src_ip]:
                ip_port_map[src_ip][dst_ip] = set()
            ip_port_map[src_ip][dst_ip].add(port)
    
    # 获取所有唯一的源IP和目的IP并排序
    all_src_ips = sorted(ip_port_map.keys())
    all_dst_ips = sorted({dst for src in ip_port_map.values() for dst in src.keys()})
    
    # 写入输出文件
    with open(output_file, 'w', encoding='utf-8') as f_out:
        for src, dst in product(all_src_ips, all_dst_ips):
            # 获取该IP对的端口列表，无通信则显示"无"
            ports = sorted(ip_port_map.get(src, {}).get(dst, []))
            port_str = ', '.join(ports) if ports else '无'
            line = f"{src} 到 {dst} 端口({port_str})，共{len(ports)}个端口\n"
            f_out.write(line)
            # 同步打印到控制台（可选）
            print(line.strip())

# 主程序入口
if __name__ == "__main__":
    log_file = input("请输入日志文件路径（需包含.csv后缀）：")
    output_file = input("请输入输出文件的保存路径及文件名：")
    process_firewall_log(log_file, output_file)
    print(f"处理完成，结果已写入{output_file}")

代码说明 #

• 数据结构优化：用嵌套字典+集合存储IP对与端口的映射，自动完成端口去重，处理10万条记录效率更高。
• 自动数据清理：批量去除IP和端口字段的前缀（如srcip=）及可能存在的引号，无需手动预处理。
• 全量组合遍历：通过itertools.product生成所有源IP与目的IP的组合，确保覆盖全部268种可能。
• 格式化输出：严格按照需求格式写入文件，同时标注每个IP对的端口数量，无通信的IP对会明确显示"无"。

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