QQ 语音通话流量分析：PCAP 解析 + IP 地理定位

抓到的 PCAP 流量包里到底藏了什么？这篇用一个实战案例——分析 QQ 语音通话的流量包，提取通信对端的 IP 地址并做地理位置定位——讲透 Python 流量分析的完整流程。

背景

QQ 语音通话采用 UDP 直连（P2P），不经过腾讯服务器中转。这意味着：

通话双方的 IP 直接暴露在流量包中
分析 PCAP 文件可以提取出通话对象的 IP 地址
配合 IP 地理位置数据库，可以定位到城市级别

应用场景：CTF 流量分析题、网络取证、安全审计。

整体流程

PCAP 文件
    │
    ├─► pyshark/scapy 读取
    │
    ├─► 过滤 UDP 数据包
    │     └─ 匹配 QQ 语音特征码 "020048"
    │
    ├─► 提取通信 IP 地址
    │
    ├─► 地理位置查询（多 API 聚合）
    │
    └─► 生成报告（JSON + Markdown）

第一步：PCAP 文件读取

Python 生态有两个主流 PCAP 处理库：

库	特点	适用场景
pyshark	基于 tshark，解析能力强，支持协议字段访问	深度协议分析
scapy	纯 Python，可直接构造/发送数据包	数据包操作、网络编程

import pyshark

def read_pcap(pcap_file):
    """读取 PCAP 文件"""
    cap = pyshark.FileCapture(
        pcap_file,
        keep_packets=False,  # 不缓存所有包（省内存）
    )
    print(f"总数据包数: {len(list(cap))}")

    # 重新创建捕获对象
    cap = pyshark.FileCapture(pcap_file)
    return cap

第二步：过滤 QQ 语音流量

QQ 语音使用 UDP 协议，数据载荷中包含固定的十六进制特征码 020048：

def is_qq_voice_packet(packet):
    """判断是否是 QQ 语音数据包"""
    try:
        # 必须是 UDP 包
        if not hasattr(packet, 'udp'):
            return False

        # 获取原始载荷（十六进制）
        if hasattr(packet.udp, 'payload'):
            payload_hex = packet.udp.payload.replace(':', '')
        else:
            return False

        # QQ 语音特征码：020048
        # 通常在 UDP 数据的前几个字节
        if '020048' in payload_hex[:100]:
            return True

        # 备选特征：端口范围
        # QQ 语音通常使用 8000 或 4000-5000 范围的端口
        src_port = int(packet.udp.srcport)
        dst_port = int(packet.udp.dstport)

        if 4000 <= src_port <= 5000 or 4000 <= dst_port <= 5000:
            if '020048' in payload_hex:
                return True

    except Exception:
        pass

    return False

为什么用 020048：这是 QQ 语音协议数据包头的固定魔数。在 Wireshark 中过滤 udp contains 02:00:48 可以直接定位到语音数据包。

第三步：提取 IP 并去重

from collections import defaultdict

def extract_ips(cap):
    """从 PCAP 中提取 QQ 语音通信的 IP 地址"""
    ip_flows = defaultdict(int)  # IP → 数据包数量

    for packet in cap:
        if not is_qq_voice_packet(packet):
            continue

        try:
            src_ip = packet.ip.src
            dst_ip = packet.ip.dst

            ip_flows[src_ip] += 1
            ip_flows[dst_ip] += 1
        except AttributeError:
            continue

    # 按数据包数量排序（包越多 → 越可能是通话对象）
    sorted_ips = sorted(ip_flows.items(), key=lambda x: x[1], reverse=True)

    return sorted_ips

第四步：IP 地理位置定位

聚合多个 IP 地理位置 API 的结果，提高准确度：

import requests

def locate_ip(ip):
    """多 API 聚合定位 IP"""
    results = []

    # API 1: ip-api.com（免费，45 次/分钟）
    try:
        resp = requests.get(f'http://ip-api.com/json/{ip}?lang=zh-CN',
                            timeout=5)
        data = resp.json()
        if data['status'] == 'success':
            results.append({
                'source': 'ip-api',
                'country': data['country'],
                'region': data['regionName'],
                'city': data['city'],
                'isp': data['isp'],
                'lat': data['lat'],
                'lon': data['lon'],
            })
    except:
        pass

    # API 2: ipgeolocation.io（更精确，免费 1000 次/天）
    try:
        api_key = 'your_api_key'
        resp = requests.get(
            f'https://api.ipgeolocation.io/ipgeo?apiKey={api_key}&ip={ip}',
            timeout=5
        )
        data = resp.json()
        results.append({
            'source': 'ipgeolocation',
            'country': data.get('country_name', ''),
            'city': data.get('city', ''),
            'isp': data.get('isp', ''),
            'lat': data.get('latitude', 0),
            'lon': data.get('longitude', 0),
        })
    except:
        pass

    # 整合结果
    if not results:
        return {'error': '所有 API 查询失败'}

    # 合并两个 API 的结果
    merged = results[0]
    if len(results) > 1:
        # 两个 API 都返回了城市，取一致的
        if merged['city'] != results[1]['city']:
            merged['city'] = f"{merged['city']}({results[1]['city']})"

    return merged

第五步：生成分析报告

JSON 详细报告

import json
from datetime import datetime

def generate_json_report(ip_flows, target_ip=None):
    """生成 JSON 格式详细报告"""
    report = {
        'analysis_time': datetime.now().isoformat(),
        'target_ip': target_ip,
        'total_flows': len(ip_flows),
        'findings': [],
    }

    for ip, packet_count in ip_flows:
        # 跳过本机 IP 和广播地址
        if ip.startswith(('127.', '10.', '192.168.', '172.16.')):
            continue

        location = locate_ip(ip)

        report['findings'].append({
            'ip': ip,
            'packet_count': packet_count,
            'location': location,
            'is_peer': packet_count > 100,  # 大流量 → 很可能是通话对端
        })

    return report

Markdown 可读摘要

def generate_markdown_summary(report):
    """生成 Markdown 可读摘要"""
    md = f"""# QQ 语音流量分析报告

**分析时间**: {report['analysis_time']}
**通信流数量**: {report['total_flows']}

## 通信对端

| IP 地址 | 数据包数 | 国家 | 城市 | ISP |
|---------|---------|------|------|-----|
"""
    for finding in report['findings']:
        loc = finding['location']
        md += f"| {finding['ip']} | {finding['packet_count']} "
        md += f"| {loc.get('country', 'N/A')} "
        md += f"| {loc.get('city', 'N/A')} "
        md += f"| {loc.get('isp', 'N/A')} |\n"

    return md

第六步：地图可视化

用 folium 在地图上标记通信对端的位置：

import folium

def visualize_on_map(report, output='qq_voice_analysis.html'):
    """在地图上可视化通信对端"""
    # 创建地图
    m = folium.Map(location=[35, 105], zoom_start=4)  # 中国中心

    for finding in report['findings']:
        loc = finding['location']
        lat = loc.get('lat', 0)
        lon = loc.get('lon', 0)

        if lat and lon:
            popup_text = f"""
            <b>IP:</b> {finding['ip']}<br>
            <b>数据包:</b> {finding['packet_count']}<br>
            <b>位置:</b> {loc.get('city')}, {loc.get('region')}<br>
            <b>ISP:</b> {loc.get('isp')}
            """

            color = 'red' if finding['is_peer'] else 'blue'
            folium.Marker(
                [lat, lon],
                popup=folium.Popup(popup_text, max_width=300),
                icon=folium.Icon(color=color, icon='info-sign'),
            ).add_to(m)

    m.save(output)
    print(f"地图已保存到 {output}")

完整分析脚本

def analyze_qq_voice_pcap(pcap_file):
    """QQ 语音流量完整分析"""
    print(f"[*] 读取 PCAP: {pcap_file}")

    # 1. 读取数据包
    cap = pyshark.FileCapture(pcap_file)
    print(f"[+] 总数据包: {len(list(cap))}")

    # 2. 提取 QQ 语音通信 IP
    cap = pyshark.FileCapture(pcap_file)  # 重新读取
    ip_flows = extract_ips(cap)
    print(f"[+] 通信 IP 数量: {len(ip_flows)}")

    # 3. 生成报告
    report = generate_json_report(ip_flows)
    with open('qq_voice_report.json', 'w', encoding='utf-8') as f:
        json.dump(report, f, ensure_ascii=False, indent=2)

    # 4. 生成 Markdown 摘要
    md = generate_markdown_summary(report)
    with open('qq_voice_summary.md', 'w', encoding='utf-8') as f:
        f.write(md)

    # 5. 地图可视化
    visualize_on_map(report)

    print("[+] 分析完成！")
    print(f"    报告: qq_voice_report.json")
    print(f"    摘要: qq_voice_summary.md")
    print(f"    地图: qq_voice_analysis.html")

实战经验

1. 大数据包文件处理

500MB+ 的 PCAP 文件不要一次性加载到内存：

# pyshark 逐包迭代（内存友好）
cap = pyshark.FileCapture('large.pcap', keep_packets=False)
for packet in cap:
    # 处理单个数据包...
    pass

2. tshark 命令行预处理

对于超大文件，先用 tshark 命令行过滤再拿 Python 分析：

1 2	# 过滤出 UDP 端口 4000-5000 的包 tshark -r huge.pcap -Y "udp.port >= 4000 and udp.port <= 5000" -w filtered.pcap

3. 特征码匹配精度

020048 可能不全出现在数据载荷头部。建议在载荷的前 200 字节内搜索，而非仅开头几个字节。