performing-network-packet-capture-analysis

执行网络数据包捕获分析

概述

网络数据包捕获（PCAP/PCAPNG 文件）代表了网络活动的最终事实来源，并提供了主机之间通信的不可辩驳的证据。PCAP 文件记录了通过网络段传输的每一个数据包，使其对于涉及数据外泄、命令与控制通信、横向移动、恶意软件投递和未经授权访问的取证调查至关重要。Wireshark 是交互式分析的主要工具，而 tshark 则为自动化处理和脚本编写提供了命令行功能。现代 PCAPNG 格式支持额外的元数据，包括接口描述、捕获注释、精确时间戳和逐数据包注释。

何时使用

• 在进行涉及执行网络数据包捕获分析的安全评估时
• 在遵循相关安全事件的事件响应程序时
• 在执行计划的安全测试或审计活动时
• 通过动手测试验证安全控制时

前提条件

• 带协议解析器的 Wireshark 4.x
• tshark 命令行工具（随 Wireshark 一起提供）
• 用于捕获和基本过滤的 tcpdump
• 带 scapy 和 pyshark 库的 Python 3.8+
• 足够的磁盘空间用于 PCAP 文件（可能达到数 GB）

捕获技术

tcpdump

# Capture all traffic on interface eth0
tcpdump -i eth0 -w capture.pcap

# Capture with rotation (100MB files, keep 10)
tcpdump -i eth0 -w capture_%Y%m%d_%H%M%S.pcap -C 100 -W 10

# Capture specific host traffic
tcpdump -i eth0 host 192.168.1.100 -w host_traffic.pcap

# Capture specific port traffic
tcpdump -i eth0 port 443 -w https_traffic.pcap

# Capture with BPF filter for suspicious ports
tcpdump -i eth0 'port 4444 or port 8080 or port 1337' -w suspicious.pcap

Wireshark 显示过滤器

# HTTP traffic
http

# DNS queries
dns

# SMB file transfers
smb2

# Specific IP communication
ip.addr == 192.168.1.100

# Failed TCP connections
tcp.flags.syn == 1 && tcp.flags.ack == 0

# Large data transfers (potential exfiltration)
tcp.len > 1000

# Specific protocol by port
tcp.port == 4444

# TLS handshakes (SNI extraction)
tls.handshake.type == 1

# HTTP POST requests
http.request.method == "POST"

# DNS queries to suspicious TLDs
dns.qry.name contains ".xyz" or dns.qry.name contains ".top"

# Beaconing detection (regular intervals)
frame.time_delta_displayed > 55 && frame.time_delta_displayed < 65

tshark 分析命令

# Extract HTTP URLs from capture
tshark -r capture.pcap -Y "http.request" -T fields -e http.host -e http.request.uri

# Extract DNS queries
tshark -r capture.pcap -Y "dns.flags.response == 0" -T fields -e dns.qry.name | sort -u

# Extract file transfers (HTTP objects)
tshark -r capture.pcap --export-objects http,exported_files/

# Extract SMB file transfers
tshark -r capture.pcap --export-objects smb,smb_files/

# Protocol hierarchy statistics
tshark -r capture.pcap -z io,phs

# Conversation statistics
tshark -r capture.pcap -z conv,tcp

# Extract TLS SNI (Server Name Indication)
tshark -r capture.pcap -Y "tls.handshake.type == 1" -T fields -e tls.handshake.extensions_server_name

# Top talkers by bytes
tshark -r capture.pcap -z endpoints,ip -q

# Extract credentials (FTP, HTTP Basic)
tshark -r capture.pcap -Y "ftp.request.command == USER || ftp.request.command == PASS || http.authorization" -T fields -e ftp.request.arg -e http.authorization

Python PCAP 分析

from scapy.all import rdpcap, IP, TCP, UDP, DNS, DNSQR, Raw
import os
import sys
import json
from collections import defaultdict, Counter
from datetime import datetime


class PCAPForensicAnalyzer:
    """Forensic analysis of PCAP files using Scapy."""

    def __init__(self, pcap_path: str, output_dir: str):
        self.pcap_path = pcap_path
        self.output_dir = output_dir
        os.makedirs(output_dir, exist_ok=True)
        self.packets = rdpcap(pcap_path)

    def get_conversations(self) -> list:
        """Extract unique IP conversations with byte counts."""
        convos = defaultdict(lambda: {"packets": 0, "bytes": 0})
        for pkt in self.packets:
            if IP in pkt:
                key = tuple(sorted([pkt[IP].src, pkt[IP].dst]))
                convos[key]["packets"] += 1
                convos[key]["bytes"] += len(pkt)

        return [
            {"src": k[0], "dst": k[1], "packets": v["packets"], "bytes": v["bytes"]}
            for k, v in sorted(convos.items(), key=lambda x: x[1]["bytes"], reverse=True)
        ]

    def extract_dns_queries(self) -> list:
        """Extract all DNS queries from the capture."""
        queries = []
        for pkt in self.packets:
            if DNS in pkt and pkt[DNS].qr == 0 and DNSQR in pkt:
                queries.append({
                    "query": pkt[DNSQR].qname.decode(errors="replace").rstrip("."),
                    "type": pkt[DNSQR].qtype,
                    "src": pkt[IP].src if IP in pkt else "unknown"
                })
        return queries

    def detect_beaconing(self, threshold_seconds: float = 5.0) -> list:
        """Detect potential beaconing activity based on regular intervals."""
        ip_timestamps = defaultdict(list)
        for pkt in self.packets:
            if IP in pkt and TCP in pkt:
                key = (pkt[IP].src, pkt[IP].dst, pkt[TCP].dport)
                ip_timestamps[key].append(float(pkt.time))

        beacons = []
        for key, times in ip_timestamps.items():
            if len(times) < 5:
                continue
            deltas = [times[i+1] - times[i] for i in range(len(times)-1)]
            if deltas:
                avg_delta = sum(deltas) / len(deltas)
                variance = sum((d - avg_delta) ** 2 for d in deltas) / len(deltas)
                if variance < threshold_seconds and avg_delta > 1:
                    beacons.append({
                        "src": key[0], "dst": key[1], "port": key[2],
                        "avg_interval": round(avg_delta, 2),
                        "variance": round(variance, 4),
                        "connection_count": len(times)
                    })
        return sorted(beacons, key=lambda x: x["variance"])

    def get_protocol_distribution(self) -> dict:
        """Get protocol distribution statistics."""
        protocols = Counter()
        for pkt in self.packets:
            if TCP in pkt:
                protocols[f"TCP/{pkt[TCP].dport}"] += 1
            elif UDP in pkt:
                protocols[f"UDP/{pkt[UDP].dport}"] += 1
        return dict(protocols.most_common(50))

    def generate_report(self) -> str:
        """Generate comprehensive PCAP analysis report."""
        report = {
            "analysis_timestamp": datetime.now().isoformat(),
            "pcap_file": self.pcap_path,
            "total_packets": len(self.packets),
            "conversations": self.get_conversations()[:50],
            "dns_queries": self.extract_dns_queries()[:200],
            "potential_beacons": self.detect_beaconing(),
            "protocol_distribution": self.get_protocol_distribution()
        }

        report_path = os.path.join(self.output_dir, "pcap_forensic_report.json")
        with open(report_path, "w") as f:
            json.dump(report, f, indent=2)

        print(f"[*] Total packets: {report['total_packets']}")
        print(f"[*] Conversations: {len(report['conversations'])}")
        print(f"[*] DNS queries: {len(report['dns_queries'])}")
        print(f"[*] Potential beacons: {len(report['potential_beacons'])}")
        return report_path


def main():
    if len(sys.argv) < 3:
        print("Usage: python process.py <pcap_file> <output_dir>")
        sys.exit(1)
    analyzer = PCAPForensicAnalyzer(sys.argv[1], sys.argv[2])
    analyzer.generate_report()


if __name__ == "__main__":
    main()

参考资料

• Wireshark 文档：https://www.wireshark.org/docs/
• PCAP 分析精通：https://insanecyber.com/mastering-pcap-review/
• SANS 网络取证：https://www.sans.org/cyber-security-courses/network-forensics/
• 用于练习的公共 PCAP：https://www.netresec.com/?page=PcapFiles