如何使用dpkt校验PCAP数据包的IP头部与UDP校验和

我现有代码以bytes形式接收PCAP数据包，已完成IP层、UDP层及私有UMS应用层的解码。希望在后续处理前检查IP头部校验和与UDP校验和，但不想手动计算校验和，需基于dpkt的解决方案。

import io
import socket
from typing import Iterator
from dpkt import pcap, ip, udp


class DecodePcap:
    """
    Receives a pcap file as bytes to decode it with dpkt.
    """
    def __init__(self, pcap_bytes: bytes) -> None:
        self.pcap = pcap.Reader(io.BytesIO(pcap_bytes))
        """
        the reader is a iterator of tuples (timestamp, buf) where timestamp is a float and buf is a bytes object.
        """
    

    def get_ums_packet_iterator(self) -> Iterator['UmsPacket']:
        """
        Returns a iterator of UmsPacket.
        """
        return map(UmsPacket, self.pcap)


class UmsPacket:
    """
    Data class to hold the fields of a packet from pcap.
    """
    def __init__(self, ums_tuple: tuple) -> None:
        self.timestamp: float = ums_tuple[0]
        ip_packet: ip.IP = ip.IP(ums_tuple[1])
        udp_packet: udp.UDP = ip_packet.data
        self.src_ip: str = socket.inet_ntoa(ip_packet.src)
        self.dst_ip: str = socket.inet_ntoa(ip_packet.dst)
        self.src_port: int = udp_packet.sport
        self.dst_port: int = udp_packet.dport
        self.data: bytes = udp_packet.data
        # check ip header checksum
        #?? assert ip_packet.sum 
        # check udp checksum
        #?? assert udp_packet.sum
        # 6th byte is the ICD version
        self.icd: int = self.data[6]

dpkt库本身内置了校验和计算方法，无需手动实现：

1. IP头部校验和检查：dpkt.ip.IP类的checksum()方法会自动计算当前IP头部的校验和，直接与数据包中的sum字段对比即可。
2. UDP校验和检查：dpkt.udp.UDP类的checksum()方法需要传入IP伪头部信息（源IP、目的IP、协议类型），因为UDP校验和的计算依赖这些内容。

修改后的完整代码如下：

import io
import socket
from typing import Iterator
from dpkt import pcap, ip, udp


class DecodePcap:
    """
    Receives a pcap file as bytes to decode it with dpkt.
    """
    def __init__(self, pcap_bytes: bytes) -> None:
        self.pcap = pcap.Reader(io.BytesIO(pcap_bytes))
        """
        the reader is a iterator of tuples (timestamp, buf) where timestamp is a float and buf is a bytes object.
        """
    

    def get_ums_packet_iterator(self) -> Iterator['UmsPacket']:
        """
        Returns a iterator of UmsPacket.
        """
        return map(UmsPacket, self.pcap)


class UmsPacket:
    """
    Data class to hold the fields of a packet from pcap.
    """
    def __init__(self, ums_tuple: tuple) -> None:
        self.timestamp: float = ums_tuple[0]
        ip_packet: ip.IP = ip.IP(ums_tuple[1])
        udp_packet: udp.UDP = ip_packet.data
        self.src_ip: str = socket.inet_ntoa(ip_packet.src)
        self.dst_ip: str = socket.inet_ntoa(ip_packet.dst)
        self.src_port: int = udp_packet.sport
        self.dst_port: int = udp_packet.dport
        self.data: bytes = udp_packet.data

        # 检查IP头部校验和
        calculated_ip_sum = ip_packet.checksum()
        assert calculated_ip_sum == ip_packet.sum, f"IP校验和不匹配：计算值{calculated_ip_sum}，数据包中值{ip_packet.sum}"

        # 检查UDP校验和，传入IP伪头部信息
        calculated_udp_sum = udp_packet.checksum(ip_packet.src, ip_packet.dst, socket.IPPROTO_UDP)
        assert calculated_udp_sum == udp_packet.sum, f"UDP校验和不匹配：计算值{calculated_udp_sum}，数据包中值{udp_packet.sum}"

        # 6th byte is the ICD version
        self.icd: int = self.data[6]

• IP校验和：ip_packet.checksum()会根据当前IP头部内容重新计算校验和，与原始数据包的sum字段对比，可验证IP头部是否被篡改或损坏。
• UDP校验和：UDP校验和计算必须包含IP伪头部，因此调用udp_packet.checksum()时必须传入源IP、目的IP和UDP协议号（用socket.IPPROTO_UDP常量更规范），否则计算结果无效。
• 若校验和不匹配，assert会直接抛出异常，你可根据业务需求调整错误处理逻辑（比如改为记录日志而非终止程序）。

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