如何使用Python过滤解析VOIP通话pcap文件中的SIP报文？

依赖准备 #

这类场景不需要从零开始写字符串切分的解析逻辑，推荐用成熟的协议解析组件减少漏判、错判，直接安装两个依赖即可：

• 核心抓包解析组件：scapy，负责读取pcap文件、识别传输层协议信息
• SIP协议扩展组件：scapy-sip，内置SIP协议全字段解析规则，自动拆分消息头、消息体边界
• 安装命令：pip install scapy scapy-sip

实现逻辑 #

核心筛选规则做双重校验，避免同端口的其他业务流量被误判为SIP：

• 第一层校验：匹配SIP常用运行端口，包括UDP/TCP 5060（明文SIP）、5061（加密SIP over TLS）
• 第二层校验：对匹配端口的载荷做格式校验，确认载荷确实以SIP请求/响应格式开头，再送入解析流程
• 解析阶段直接调用内置SIP解析规则，自动提取所有消息头字段、识别消息体边界，不需要手动按\r\n\r\n切分文本

可直接运行的参考代码 #

from scapy.all import rdpcap, UDP, TCP, Raw
from scapy_sip import SIP

# 替换为你自己的pcap文件本地路径
PCAP_FILE_PATH = "./voip_capture.pcap"
# 兼容SIP常用运行端口，自定义端口可直接添加到集合中
SIP_MATCH_PORTS = {5060, 5061}

def parse_single_sip_packet(pkt):
    # 第一步：先做传输层端口匹配
    is_sip_transport = False
    transport_layer = None
    if pkt.haslayer(UDP):
        transport_layer = pkt[UDP]
        if transport_layer.sport in SIP_MATCH_PORTS or transport_layer.dport in SIP_MATCH_PORTS:
            is_sip_transport = True
    elif pkt.haslayer(TCP):
        transport_layer = pkt[TCP]
        if transport_layer.sport in SIP_MATCH_PORTS or transport_layer.dport in SIP_MATCH_PORTS:
            is_sip_transport = True
    
    if not is_sip_transport:
        return None

    # 第二步：校验载荷合法性，做SIP解析
    try:
        raw_load = bytes(pkt[Raw].load)
        sip_pkt = SIP(raw_load)
    except Exception:
        # 载荷不是合法SIP格式的直接跳过
        return None

    # 提取全量SIP消息头
    extracted_headers = {}
    for field in sip_pkt.fields_desc:
        if field.name in sip_pkt.fields:
            extracted_headers[field.name] = sip_pkt.fields[field.name]

    # 提取SIP消息体（无消息体时返回空字符串）
    sip_body = sip_pkt.content.decode("utf-8", errors="ignore") if sip_pkt.content else ""

    return {
        "传输来源": f"{pkt.src}:{transport_layer.sport}",
        "传输目的": f"{pkt.dst}:{transport_layer.dport}",
        "消息头全量字段": extracted_headers,
        "消息体内容": sip_body
    }

if __name__ == "__main__":
    # 加载全量pcap报文
    all_packets = rdpcap(PCAP_FILE_PATH)
    parsed_sip_list = []
    for pkt in all_packets:
        sip_result = parse_single_sip_packet(pkt)
        if sip_result:
            parsed_sip_list.append(sip_result)
    
    # 输出所有解析到的SIP报文
    for index, sip_msg in enumerate(parsed_sip_list, 1):
        print(f"\n======= 第{index}条SIP报文 =======")
        print(f"传输链路: {sip_msg['传输来源']} -> {sip_msg['传输目的']}")
        print("--- SIP消息头 ---")
        for header_key, header_val in sip_msg["消息头全量字段"].items():
            print(f"{header_key}: {header_val}")
        print("--- SIP消息体 ---")
        print(sip_msg["消息体内容"])

常见注意事项 #

• 如果抓包中包含5061端口的SIP over TLS加密流量，没有对应TLS会话密钥的前提下，无法直接解析出明文SIP内容，这类包会被脚本自动跳过
• 部分运营商或企业部署的SIP业务会使用非标准自定义端口，可将对应端口号直接加入SIP_MATCH_PORTS集合，配合载荷格式校验即可正常识别
• 解码消息体时统一加errors="ignore"容错逻辑，避免遇到非UTF-8编码的二进制载荷导致脚本运行中断

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