营销网站建站seo内容优化
引言
在网络安全和流量管理中，解析网络协议数据包是了解网络行为和检测潜在威胁的关键步骤。本文介绍了如何使用Python解析和分析TCP、UDP和ICMP协议的数据包，并统计端口的访问次数。本文的示例代码展示了如何处理不同协议的数据包，提取关键信息，并对网络流量进行深入分析。
环境设置
在开始之前，我们需要设置环境以便使用pyshark库进行数据包捕获和解析。首先，我们需要安装Wireshark并确保tshark命令行工具可用。然后，我们在代码中设置tshark的路径，并创建一个用于保存下载文件的目录。
import os# 设置tshark路径
os.environ['TSHARK_PATH'] = 'D:\\Wireshark\\tshark.exe'
os.environ['PATH'] += os.pathsep + 'D:\\Wireshark'# 创建下载目录
DOWNLOAD_DIR = 'xiazai'
if not os.path.exists(DOWNLOAD_DIR):os.makedirs(DOWNLOAD_DIR)
数据包解析
我们的核心函数是analyze_pcap，它使用pyshark库解析给定的PCAP文件，并提取TCP、UDP和ICMP协议的数据包。通过该函数，我们能够捕获流量中的源IP、目的IP、源端口、目的端口等信息，并识别DNS查询和TLS握手数据。
import pyshark
from collections import defaultdictdef analyze_pcap(pcap_file, tshark_path):cap = pyshark.FileCapture(pcap_file, tshark_path=tshark_path)flows = defaultdict(list)dns_queries = defaultdict(list)tls_handshakes = defaultdict(list)icmp_packets = []for packet in cap:try:src_ip = packet.ip.srcdst_ip = packet.ip.dsttransport_layer = packet.transport_layerif transport_layer in ['TCP', 'UDP']:src_port = packet[transport_layer].srcportdst_port = packet[transport_layer].dstportflow_key = (src_ip, dst_ip, src_port, dst_port)flows[flow_key].append(packet)if transport_layer == 'UDP' and hasattr(packet, 'dns'):dns_query = packet.dns.qry_namedns_queries[flow_key].append(dns_query)if hasattr(packet, 'tls'):if hasattr(packet.tls, 'handshake_type'):tls_handshake_type = packet.tls.handshake_typetls_handshakes[flow_key].append(tls_handshake_type)elif transport_layer == 'ICMP':icmp_packets.append(packet)except AttributeError:continuereturn flows, dns_queries, tls_handshakes, icmp_packets
数据包解析
我们的核心函数是analyze_pcap，它使用pyshark库解析给定的PCAP文件，并提取TCP、UDP和ICMP协议的数据包。通过该函数，我们能够捕获流量中的源IP、目的IP、源端口、目的端口等信息，并识别DNS查询和TLS握手数据。
import pyshark
from collections import defaultdictdef analyze_pcap(pcap_file, tshark_path):cap = pyshark.FileCapture(pcap_file, tshark_path=tshark_path)flows = defaultdict(list)dns_queries = defaultdict(list)tls_handshakes = defaultdict(list)icmp_packets = []for packet in cap:try:src_ip = packet.ip.srcdst_ip = packet.ip.dsttransport_layer = packet.transport_layerif transport_layer in ['TCP', 'UDP']:src_port = packet[transport_layer].srcportdst_port = packet[transport_layer].dstportflow_key = (src_ip, dst_ip, src_port, dst_port)flows[flow_key].append(packet)if transport_layer == 'UDP' and hasattr(packet, 'dns'):dns_query = packet.dns.qry_namedns_queries[flow_key].append(dns_query)if hasattr(packet, 'tls'):if hasattr(packet.tls, 'handshake_type'):tls_handshake_type = packet.tls.handshake_typetls_handshakes[flow_key].append(tls_handshake_type)elif transport_layer == 'ICMP':icmp_packets.append(packet)except AttributeError:continuereturn flows, dns_queries, tls_handshakes, icmp_packets
文件提取和分析
在捕获流量之后，我们需要提取并分析其中的文件和图像。以下代码展示了如何从HTTP流量中提取ZIP、PDF和图像文件，并检测其中的敏感信息或潜在威胁。
import zipfile
import PyPDF2
from PIL import Image
import re
from datetime import datetimedef extract_files(flows, keywords):# 省略代码passdef check_and_extract_zip(file_name, keywords):# 省略代码passdef check_and_extract_pdf(file_name, keywords):# 省略代码passdef check_and_extract_image(file_name):# 省略代码passdef detect_sql_injection(payload):# 省略代码passdef clean_text(text):# 省略代码passdef save_sql_injection_data(flow, post_data, pcap_file_name):# 省略代码passdef save_password_post_data(flow, post_data, pcap_file_name):# 省略代码pass
应用程序和端口分析
为了更全面地了解网络流量，我们还需要分析端口的使用情况和应用层协议的数据。以下代码展示了如何统计各端口的访问次数，并检测潜在的SQL注入攻击和敏感信息泄露。
def detect_application(flows, dns_queries, tls_handshakes, icmp_packets, pcap_file_name):flow_summary = defaultdict(int)port_usage = defaultdict(lambda: defaultdict(int))src_dst_port_usage = defaultdict(lambda: defaultdict(lambda: defaultdict(int)))password_patterns = [r"password", r"passwd", r"pwd", r"pass"]>>>>关注加好友 发源代码<<<< print("\nHTTP流量检测:")for flow_key, count in flow_summary.items():host, method, url = flow_keyprint(f"主机: {host}, 请求方法: {method}, URL: {url} - 出现次数: {count}")print("\n端口使用情况:")for port, methods in port_usage.items():for method, count in methods.items():print(f"端口 {port} 上的 {method} 请求次数: {count}")print("\n源IP到目的IP的端口使用情况:")for src_ip, dst_ips in src_dst_port_usage.items():for dst_ip, ports in dst_ips.items():for port, count in ports.items():print(f"{src_ip} -> {dst_ip} 端口 {port} 使用次数: {count}")print("\nDNS查询:")for flow_key, queries in dns_queries.items():src_ip, dst_ip, src_port, dst_port = flow_keyfor query in queries:print(f"{src_ip} -> {dst_ip} : DNS查询 {query}")print("\nTLS握手:")for flow_key, handshakes in tls_handshakes.items():src_ip, dst_ip, src_port, dst_port = flow_keyfor handshake in handshakes:print(f"{src_ip} -> {dst_ip} : TLS握手类型 {handshake}")print("\nICMP包:")for packet in icmp_packets:src_ip = packet.ip.srcdst_ip = packet.ip.dsticmp_type = packet.icmp.typeicmp_code = packet.icmp.codeprint(f"ICMP包：{src_ip} -> {dst_ip}, 类型: {icmp_type}, 代码: {icmp_code}")
主程序
在主程序中，我们首先调用analyze_pcap函数解析PCAP文件，然后调用其他函数进行文件提取、应用程序检测和端口分析。
if __name__ == '__main__':pcap_file = '.\\test3.pcapng'tshark_path = 'D:\\Wireshark\\tshark.exe'keywords = ['password', 'secret', 'confidential']try:flows, dns_queries, tls_handshakes, icmp_packets = analyze_pcap(pcap_file, tshark_path)extract_files(flows, keywords)detect_application(flows, dns_queries, tls_handshakes, icmp_packets, os.path.basename(pcap_file))except FileNotFoundError:print(f"文件未找到: {pcap_file}")except Exception as e:print(f"发生错误: {e}")
运行结果
作者ps：对于流量分析我暂时只能写到这个部分如果你有更好的想法可以和我私聊，这个程序也是我之间打打ctf玩玩也能。