如何用Python+Scapy分析PCAP文件的TCP会话完整性？

核心思路 #

呼叫会话的完整性本质是双向数据包交互：既要存在发起方到接收方的请求/媒体包，也要有接收方到发起方的响应/媒体包。我们可以通过标准化五元组（源IP、源端口、目的IP、目的端口、协议）将正向/反向流归为同一个会话，再筛选出双向均有数据包的会话；如果是VoIP呼叫（如SIP协议），还可以针对性校验信令交互的完整性（如INVITE请求+200 OK响应）。

实现代码 #

1. 基础版本：识别所有双向交互会话

from scapy.all import rdpcap, IP, TCP, UDP

def get_standardized_flow(pkt):
    """生成标准化的会话键，确保正向/反向流对应同一个会话"""
    if IP not in pkt:
        return None
    src_ip = pkt[IP].src
    dst_ip = pkt[IP].dst
    # 提取端口（仅TCP/UDP协议）
    src_port = pkt.sport if (TCP in pkt or UDP in pkt) else 0
    dst_port = pkt.dport if (TCP in pkt or UDP in pkt) else 0
    proto = pkt[IP].proto
    
    # 按IP+端口排序，统一正向/反向流的键格式
    if (src_ip, src_port) < (dst_ip, dst_port):
        return (src_ip, src_port, dst_ip, dst_port, proto)
    else:
        return (dst_ip, dst_port, src_ip, src_port, proto)

def extract_complete_bidirectional_sessions(pcap_path):
    sessions = {}
    pkts = rdpcap(pcap_path)
    
    for pkt in pkts:
        flow_key = get_standardized_flow(pkt)
        if not flow_key:
            continue
        
        # 判断当前数据包的方向（正向/反向）
        is_forward = (pkt[IP].src == flow_key[0] and 
                      (pkt.sport == flow_key[1] if (TCP in pkt or UDP in pkt) else True))
        
        # 初始化会话记录
        if flow_key not in sessions:
            sessions[flow_key] = {
                "forward_pkts": 0,
                "reverse_pkts": 0,
                "all_pkts": []
            }
        
        # 更新计数和数据包列表
        if is_forward:
            sessions[flow_key]["forward_pkts"] += 1
        else:
            sessions[flow_key]["reverse_pkts"] += 1
        sessions[flow_key]["all_pkts"].append(pkt)
    
    # 筛选出双向均有数据包的完整会话
    complete_sessions = {
        k: v for k, v in sessions.items()
        if v["forward_pkts"] > 0 and v["reverse_pkts"] > 0
    }
    return complete_sessions

# 调用示例
if __name__ == "__main__":
    pcap_file = "your_call_traffic.pcap"
    complete_sessions = extract_complete_bidirectional_sessions(pcap_file)
    
    # 输出结果示例
    for flow, data in complete_sessions.items():
        print(f"标准化会话五元组: {flow}")
        print(f"正向包数: {data['forward_pkts']}, 反向包数: {data['reverse_pkts']}\n")

2. 进阶版本：针对SIP呼叫的完整性校验

如果是VoIP呼叫场景，可进一步校验SIP信令的完整交互（如INVITE请求+200 OK响应），确保是真正完成的呼叫会话：

def is_valid_sip_call(session_data):
    """检查会话中是否包含完整的SIP呼叫信令"""
    has_invite = False
    has_ok_response = False
    
    for pkt in session_data["all_pkts"]:
        if "SIP" in pkt:
            sip_layer = pkt["SIP"]
            if sip_layer.method == "INVITE":
                has_invite = True
            elif sip_layer.status_code == "200":
                has_ok_response = True
    
    return has_invite and has_ok_response

# 提取完整的SIP呼叫会话
sip_complete_calls = {
    k: v for k, v in complete_sessions.items()
    if is_valid_sip_call(v)
}

# 输出SIP呼叫详情
for flow, data in sip_complete_calls.items():
    print(f"SIP呼叫会话: {flow}")
    for pkt in data["all_pkts"]:
        if "SIP" in pkt:
            sip = pkt["SIP"]
            print(f"  SIP {sip.method if hasattr(sip, 'method') else f'响应 {sip.status_code}'}: {sip.CSeq}")

关键说明 #

• 标准化五元组：解决了正向流（A→B）和反向流（B→A）的会话合并问题，确保同一呼叫的双向数据被归为同一个会话。
• 扩展性：如果是RTP媒体流，可基于SSRC字段进一步关联SIP信令和媒体流，完善呼叫会话的完整性校验。
• 性能优化：如果PCAP文件过大，可使用rdpcap(pcap_path, count=N)分批读取，或使用Scapy的sniff()函数实时处理。

内容的提问来源于stack exchange，提问作者Somanath Nayak