网际校验和算法
摘 要
本文旨在研究和实现网际校验和（Internet Checksum）算法。通过阅读《RFC 1071》文档理解该算法的工作原理，并使用编程语言实现网际校验和的计算过程。本项目将对不同类型的网络报文（包括ICMP、TCP、UDP等）进行差错检验，对捕获的报文进行校验和计算并验证其正确性。项目包括报文数据的读取、校验和的计算以及结果的输出展示等功能。本报告详细描述了算法的实现过程、测试方法和实验结果，并对网际校验和算法在网络通信中的应用进行了分析。通过本次探研，加深了对网络协议差错检测机制的理解，提高了协议分析与实现的能力。
关键字：网际校验和、差错检测、网络协议、报文验证
一、算法概述
网际校验和算法是一种简单且高效的错误检测机制，广泛应用于网络协议中，如IP、TCP、UDP等。其核心思想是在数据传输前生成一个校验和，并通过对数据进行分段求和及取反操作来实现错误检测。接收端通过相同的算法重新计算校验和，如果结果为全0，则说明数据传输过程中未发生错误。
1 算法原理
网际校验和算法由两个核心组件构成：校验和的生成与验证。在数据传输过程中，发送方负责生成校验和，而接收方则执行校验和的验证工作。下面是校验和的详细生成流程：
①数据分块：在算法开始时，首先将数据流划分为16位（2字节）为单位的字块。如果数据总长度不是16位的整数倍，则需要在末尾填充0来补足。
②把校验和字段置为0：在计算校验和之前，需要先将校验和字段的值置为0。这是因为校验和字段本身也会参与计算，但计算时应当使用0值。
③逐块求和：对每个16位字块进行加和，若产生进位（即求和结果超过16位，产生大于65535的结果），则将进位加回到结果的最低位。这个进位的处理是算法的关键，使得加法操作能避免丢失信息。
④按位取反：将上述求和结果按位取反（即对所有位进行反转），得到最终的校验和。取反操作是为了增强错误检测的能力，避免某些类型的错误模式（如全0或全1错误）无法被检测到。
校验和的验证则在接收端进行。在接收端，重新进行相同的求和和取反操作（不进行第②步）。如果最终结果是全0，则认为数据未发生错误；如果结果不是全0，则说明数据在传输过程中出现了错误。
该算法的设计使其能够快速且有效地检测常见的数据传输错误，如单比特错误、字节错误等。但它也有局限性，对于一些特定类型的错误（例如数据的字块顺序被改变）可能不敏感，因此它并不是一种完美的错误检测算法，但在多数应用场景下，已能满足需求。
2 算法特点
网际校验和算法在计算机网络中扮演着重要角色，其设计目标是通过简单的操作实现数据传输中的错误检测。由于其独特的特点和广泛的应用价值，该算法在多种网络协议中被广泛采用。以下将从多个方面介绍网际校验和算法的主要特点：
（1）简单高效：计算过程中仅需基本的加法和按位取反操作，执行效率较高，适合硬件加速实现，适用于实时性要求较高的应用场景。
（2）检错能力适中：能有效检测大多数单比特错误和字块错误，但对某些复杂的错误模式（如字节重排）敏感度较低。
（3）广泛应用：被广泛用于IP、TCP、UDP等网络协议中，尤其在需要高效和实时处理的环境中，已成为标准校验方法之一。
（4）实现灵活：由于其简单性，校验和的计算可以根据数据长度和类型的不同进行调整，灵活性较高。
3 算法应用
网际校验和算法因其独特的技术特征，在实际应用中具有广泛的实用价值。该算法凭借计算简便、检错能力适度等优势，已被广泛集成至各类网络协议和应用系统中，为数据传输的完整性与可靠性提供了有效保障。以下将详细介绍该算法在不同领域中的典型应用：
（1）网络通信协议：
- IP层（IPv4首部校验和）：用于检测IP数据包的头部是否在传输过程中被篡改或损坏。
- 传输层（TCP、UDP报文段校验和）：用于校验TCP和UDP报文段中的数据完整性，确保数据在传输过程中未被修改。
（2）数据完整性验证：
- 文件传输与存储完整性检测：校验和可用于验证文件传输过程中的数据完整性，确保文件未在传输中发生损坏。
- 网络设备数据包处理：网络设备通过校验和算法确保接收和转发的数据包完整性。
（3）嵌入式系统：
在嵌入式系统中，尤其是通信模块，使用该算法进行数据校验，确保通信过程中的数据不被篡改或丢失。
（4）网络安全与监控：
网络入侵检测与数据包分析工具：在安全系统中，算法可用于检测网络数据包中的错误或恶意篡改，从而保障网络的安全性。
二、编程实现
1 程序编写
下文将详细介绍如何通过编程实现网际校验和算法。本章节将从编程环境搭建开始，逐步说明代码实现的具体步骤、关键函数的设计思路以及数据处理的方法。通过这个实践过程，我们不仅能够深入理解网际校验和算法的工作原理，也能掌握如何将理论算法转化为可执行的程序代码。程序采用Python语言实现，代码结构清晰，便于理解和后续的维护与扩展。
1.1 编程环境
编程环境的正确配置是确保程序开发和运行顺利进行的重要前提。我选择了Windows 11作为开发平台，搭配功能强大的PyCharm作为集成开发环境，使用广受欢迎的Python编程语言，并将处理来自Wireshark工具导出的“.txt”格式数据文件。这些工具的选择既考虑了开发效率，也兼顾了程序的可移植性和扩展性。
- 操作系统：Windows 11
- 开发工具：PyCharm
- 编程语言：Python
- 数据文档格式：从Wireshark工具软件导出的原始数据包，为“.txt”格式。
图1 数据文档格式图
1.2 校验流程
本次编码不仅完成了对IP数据报首部的校验，还实现了ICMP、TCP、UDP报文的校验功能。鉴于整个实现流程较为复杂，此处仅展示IP数据报首部的关键校验流程。具体的IP数据报首部校验流程如下图所示（非标准流程图）：
图2 IP数据报首部校验流程图
1.3 核心代码
鉴于IP报文首部校验和的计算与验证机制涉及多个分散的代码模块，此处仅展示其核心实现部分。完整的实现细节请参阅源代码。
（1）校验和计算函数`checksum_calculating`：实现标准的网际校验和算法，用于计算IP头部校验和。在函数中，校验和的计算流程如下：①从输入数据中检查长度，若为奇数则进行字节补齐；②对数据按16位为单位进行累加计算；③将进位加至最低位；④对最终结果取反得到校验和。具体代码如下图所示：
图3 校验和计算函数
（2）校验和验证函数`process_packet_to_string`：实现接收数据包的校验和验证。在函数中，校验和的验证流程如下：①从接收的IP首部中提取原始校验和；②构造用于校验和计算的IP首部数据；③调用`checksum_calculating`函数计算当前校验和；④将计算得到的校验和与原始校验和进行比对。部分代码如下图所示：
图4 校验和验证函数（部分）
2 程序测试
为验证程序的正确性，下面我将对我编写的用于网际校验和验证的程序进行测试。
2.1 测试文件
本次测试的数据文件均来自于计网实验5，从中选取典型的ICMP、UDP、TCP报文各一个。测试文件分别为“ICMP.txt”、“TCP.txt”、“UDP.txt”。这些文件包含了真实网络环境中的典型数据包，能够很好地验证算法在实际应用场景中的表现。测试文件内容格式如前文所述。
图5 测试文件
2.2 测试结果
（1）ICMP
下图是ICMP报文的具体内容，可以看到它的IP首部校验和为“f2 24”（图中蓝色高亮部分）。
图6 ICMP报文
下图是程序读取该报文后的测试结果，程序运行结果正确，与预期一致。
图7 ICMP报文测试结果
（2）TCP
下图是TCP报文的具体内容，可以看到它的IP首部校验和为“cd 6c”（图中蓝色高亮部分）。
图8 TCP报文
下图是程序读取该报文后的测试结果，程序运行结果正确，与预期一致。
图9 TCP报文测试结果
（3）UDP
下图是UDP报文的具体内容，可以看到它的IP首部校验和为“c0 d3”（图中蓝色高亮部分）。
图10 UDP报文
下图是程序读取该报文后的测试结果，程序运行结果正确，与预期一致。
图11 UDP报文测试结果
结 论
本次探研通过深入分析网际校验和算法的原理并进行实践实现，取得了以下主要成果：
（1）完成了网际校验和算法的理论研究和代码实现。通过研读《RFC 1071》文档，深入理解了该算法的工作原理，包括数据分块、校验和计算和验证等核心过程。研究表明该算法具有实现简单、计算效率高等特点，这也解释了其在网络协议中的广泛应用。
（2）成功开发了一个能够处理多种网络协议报文的校验程序。该程序不仅实现了IP数据报首部的校验，还扩展支持了ICMP、TCP、UDP等多种协议报文的校验功能。通过实际测试，程序能够正确计算和验证各类报文的校验和，验证结果与实际数据包中的校验和完全匹配。
（3）通过实践验证了网际校验和算法在差错检测方面的有效性。虽然该算法在某些特定错误模式（如字节重排）的检测上存在局限性，但其简单高效的特点使其非常适合网络通信中的实时差错检测需求。
（4）本次探研的实践过程加深了对网络协议差错检测机制的理解，提高了协议分析与实现的能力。通过编程实现和测试验证，不仅掌握了算法的技术细节，也认识到了在实际网络环境中确保数据传输可靠性的重要性。
总的来说，本次探研不仅完成了对网际校验和算法的理论学习和实践实现，还通过具体的程序开发和测试验证了该算法的实用价值。这些工作为进一步理解和应用网络协议中的差错检测机制奠定了基础。
参考文献
- Braden R, Borman D A, Partridge C. RFC 1071 Computing the Internet Checksum[S]. Fremont: RFC Editor, 1988.
- 李毅,张帆,张润宇.IPv4头部校验和的反码算法[J].武汉理工大学学报,2003,(04):64-68.
- 刘派.IP首部校验算法[J].电脑知识与技术,2010,6(19):5194-5196.
- 乔世成,张智丰,廉洁.IP首部校验和算法研究[J].内蒙古民族大学学报(自然科学版),2016,31(05):400-402.DOI:10.14045/j.cnki.15-1220.2016.05.010.
- 孔庆春.当前应用于计算机通信中的差错检测与控制技术[J].信息与电脑(理论版),2017,(18):146-148.
参考代码
import struct
import tkinter as tk
from tkinter import filedialog, messagebox, scrolledtext
def checksum_calculating(data):
"""
计算给定数据的校验和（使用Internet Checksum算法）。
参数:
data (bytes): 要计算校验和的字节数据。
返回:
int: 计算得到的校验和。
"""
# 如果数据长度为奇数，补一个0字节
if len(data) % 2 != 0:
data += b'\x00'
checksum = 0
for i in range(0, len(data), 2):
word = struct.unpack('!H', data[i:i + 2])[0]
checksum += word
# 处理进位
while checksum >> 16:
checksum = (checksum & 0xffff) + (checksum >> 16)
checksum = ~checksum & 0xffff
return checksum
def parse_ethernet_frame(data):
"""
解析以太网帧头，提取EtherType。
参数:
data (bytes): 以太网帧的字节数据。
返回:
tuple: (ethertype (int), payload (bytes))。
异常:
ValueError: 数据长度不足以解析以太网帧头。
"""
if len(data) < 14:
raise ValueError("数据长度不足以解析以太网帧头")
# 查找EtherType为0x0800（IPv4）的位置
for i in range(len(data) - 2):
if data[i] == 0x08 and data[i + 1] == 0x00:
eth_header = data[:i + 2]
payload = data[i + 2:]
return 0x0800, payload
# 如果未找到，使用标准解析方法
eth_header = data[:14]
eth_fields = struct.unpack('!6s6sH', eth_header)
ethertype = eth_fields[2]
payload = data[14:]
return ethertype, payload
def parse_ip_packet(data):
"""
解析IP数据包，提取协议类型和IP头部。
参数:
data (bytes): IP数据包的字节数据。
返回:
tuple: (protocol (int), ip_header (bytes), payload (bytes))。
异常:
ValueError: 数据长度不足以解析IP头部或完整的IP头部。
"""
if len(data) < 20:
raise ValueError("数据长度不足以解析IP头部")
version_ihl = data[0]
ihl = (version_ihl & 0x0F) * 4 # 头部长度
if len(data) < ihl:
raise ValueError("数据长度不足以解析完整的IP头部")
ip_header = data[:ihl]
protocol = data[9]
payload = data[ihl:]
return protocol, ip_header, payload
def parse_icmp_packet(data):
"""
解析ICMP数据包。
参数:
data (bytes): ICMP数据包的字节数据。
返回:
tuple: (icmp_type (int), icmp_code (int), icmp_data (bytes),
received_checksum (int), checksum_data (bytes))。
异常:
ValueError: 数据长度不足以解析ICMP头部。
"""
if len(data) < 4:
raise ValueError("数据长度不足以解析ICMP头部")
icmp_header = data[:4]
icmp_fields = struct.unpack('!BBH', icmp_header)
icmp_type = icmp_fields[0]
icmp_code = icmp_fields[1]
received_checksum = icmp_fields[2]
icmp_data = data[4:]
# 创建用于计算校验和的数据（校验和字段置0）
checksum_data = data[:2] + b'\x00\x00' + data[4:]
return icmp_type, icmp_code, icmp_data, received_checksum, checksum_data
def parse_tcp_packet(data):
"""
解析TCP数据包。
参数:
data (bytes): TCP数据包的字节数据。
返回:
tuple: (src_port (int), dest_port (int), seq_num (int), ack_num (int),
tcp_header_for_checksum (bytes), tcp_data (bytes),
received_checksum (int))。
异常:
ValueError: 数据长度不足以解析TCP头部。
"""
if len(data) < 20:
raise ValueError("数据长度不足以解析TCP头部")
tcp_header = data[:20]
tcp_fields = struct.unpack('!HHLLHHHH', tcp_header)
src_port = tcp_fields[0]
dest_port = tcp_fields[1]
seq_num = tcp_fields[2]
ack_num = tcp_fields[3]
offset_reserved_flags = tcp_fields[4]
window = tcp_fields[5]
received_checksum = tcp_fields[6]
urgent_pointer = tcp_fields[7]
# 提取数据偏移、保留和标志位
data_offset = (offset_reserved_flags >> 12) & 0xF # 数据偏移（高4位）
reserved = (offset_reserved_flags >> 6) & 0x3F # 保留位（中间6位）
flags = offset_reserved_flags & 0x3F # 标志位（低6位）
data_offset = data_offset * 4
tcp_data = data[data_offset:]
# 校验和计算时，将校验和字段置零，但保留整个20字节头部
tcp_header_for_checksum = tcp_header[:16] + b'\x00\x00' + tcp_header[18:20]
return src_port, dest_port, seq_num, ack_num, tcp_header_for_checksum, tcp_data, received_checksum
def parse_udp_packet(data):
"""
解析UDP数据包。
参数:
data (bytes): UDP数据包的字节数据。
返回:
tuple: (src_port (int), dest_port (int), length (int),
udp_header_for_checksum (bytes), udp_data (bytes),
received_checksum (int))。
异常:
ValueError: 数据长度不足以解析UDP头部。
"""
if len(data) < 8:
raise ValueError("数据长度不足以解析UDP头部")
udp_header = data[:8]
udp_fields = struct.unpack('!HHHH', udp_header)
src_port = udp_fields[0]
dest_port = udp_fields[1]
length = udp_fields[2]
received_checksum = udp_fields[3]
udp_data = data[8:]
# 将校验和字段置零用于计算
udp_header_for_checksum = udp_header[:6] + b'\x00\x00'
return src_port, dest_port, length, udp_header_for_checksum, udp_data, received_checksum
def hexstr_to_bytes(line):
"""
将十六进制字符串转换为字节数据。
参数:
line (str): 包含十六进制数的字符串，每个字节由两个十六进制字符表示，
字节之间用'|'分隔。
返回:
bytes or None: 转换后的字节数据，如果转换失败则返回None。
"""
try:
parts = line.strip().split('|')
# 去除空白部分
parts = [p.strip() for p in parts if p.strip()]
if len(parts) > 1:
parts = parts[1:] # 去掉偏移量字段
# 确保每个部分有两个字符，不足则前置补零
hex_pairs = [p.zfill(2) for p in parts]
hex_cleaned = ''.join(hex_pairs)
return bytes.fromhex(hex_cleaned)
except Exception:
return None
def process_packet_to_string(packet_bytes):
"""
处理数据包并将解析结果格式化为字符串。
参数:
packet_bytes (bytes): 要处理的数据包字节数据。
返回:
str: 格式化后的解析结果。
"""
result = []
result.append(f"原始报文数据 ({len(packet_bytes)} 字节): {packet_bytes.hex()}")
try:
# 解析以太网帧
ethertype, payload = parse_ethernet_frame(packet_bytes)
result.append(f"以太网类型: 0x{ethertype:04x}")
if ethertype == 0x0800: # IPv4
protocol, ip_header, ip_payload = parse_ip_packet(payload)
result.append(f"IP协议版本: {ip_header[0] >> 4}")
result.append(f"IP首部长度: {(ip_header[0] & 0x0F) * 4} 字节")
result.append(f"IP协议: {protocol}")
# IP校验和
received_ip_checksum = struct.unpack('!H', ip_header[10:12])[0]
ip_header_for_checksum = ip_header[:10] + b'\x00\x00' + ip_header[12:]
calculated_ip_checksum = checksum_calculating(ip_header_for_checksum)
result.append(f"接收的IP校验和: 0x{received_ip_checksum:04x}")
result.append(f"计算的IP校验和: 0x{calculated_ip_checksum:04x}")
if received_ip_checksum == calculated_ip_checksum:
result.append("IP校验和正确")
else:
result.append("IP校验和错误")
if protocol == 1: # ICMP
icmp_type, icmp_code, icmp_data, received_checksum, checksum_data = parse_icmp_packet(ip_payload)
calculated_checksum = checksum_calculating(checksum_data)
result.append(f"ICMP类型: {icmp_type}, 代码: {icmp_code}")
result.append(f"ICMP接收的校验和: 0x{received_checksum:04x}")
result.append(f"ICMP计算的校验和: 0x{calculated_checksum:04x}")
if received_checksum == calculated_checksum:
result.append("ICMP校验和正确")
else:
result.append("ICMP校验和错误")
elif protocol == 6: # TCP
src_port, dest_port, seq_num, ack_num, tcp_header_for_checksum, tcp_data, received_checksum = parse_tcp_packet(
ip_payload)
source_ip = ip_header[12:16] # 源IP地址
destination_ip = ip_header[16:20] # 目的IP地址
reserved_zero = b'\x00'
protocol_byte = struct.pack('!B', protocol)
segment_length = len(tcp_header_for_checksum) + len(tcp_data)
tcp_length = struct.pack('!H', segment_length)
# 构造伪头部用于校验和计算
pseudo_header = source_ip + destination_ip + reserved_zero + protocol_byte + tcp_length
checksum_data = pseudo_header + tcp_header_for_checksum + tcp_data
if len(checksum_data) % 2 != 0:
checksum_data += b'\x00'
calculated_checksum = checksum_calculating(checksum_data)
result.append(f"TCP源端口: {src_port}, 目的端口: {dest_port}")
result.append(f"TCP接收的校验和: 0x{received_checksum:04x}")
result.append(f"TCP计算的校验和: 0x{calculated_checksum:04x}")
if received_checksum == calculated_checksum:
result.append("TCP校验和正确")
else:
result.append("TCP校验和错误")
elif protocol == 17: # UDP
src_port, dest_port, length, udp_header_for_checksum, udp_data, received_checksum = parse_udp_packet(
ip_payload)
source_ip = ip_header[12:16] # 源IP地址
destination_ip = ip_header[16:20] # 目的IP地址
reserved_zero = b'\x00'
protocol_byte = struct.pack('!B', protocol)
udp_length = struct.pack('!H', length)
# 构造伪头部用于校验和计算
pseudo_header = source_ip + destination_ip + reserved_zero + protocol_byte + udp_length
checksum_data = pseudo_header + udp_header_for_checksum + udp_data
if len(checksum_data) % 2 != 0:
checksum_data += b'\x00'
calculated_checksum = checksum_calculating(checksum_data)
result.append(f"UDP源端口: {src_port}, 目的端口: {dest_port}")
result.append(f"UDP接收的校验和: 0x{received_checksum:04x}")
result.append(f"UDP计算的校验和: 0x{calculated_checksum:04x}")
if received_checksum == calculated_checksum:
result.append("UDP校验和正确")
else:
result.append("UDP校验和错误")
else:
result.append(f"未支持的IP协议类型: {protocol}")
else:
result.append(f"未支持的以太网类型: 0x{ethertype:04x}")
except ValueError as ve:
result.append("解析报文时出错: " + str(ve))
return '\n'.join(result)
def main_gui():
"""
主函数，创建图形界面，选择文件，解析报文，计算校验和并显示结果。
"""
root = tk.Tk()
root.title("网际校验和分析工具")
root.geometry("800x600")
# 创建可滚动文本区域显示结果
text_area = scrolledtext.ScrolledText(root, wrap=tk.WORD, width=100, height=40)
text_area.pack(padx=10, pady=10, fill=tk.BOTH, expand=True)
def select_file():
"""
处理文件选择和报文解析。
"""
file_path = filedialog.askopenfilename(
title="选择测试文件",
filetypes=(("文本文件", "*.txt"), ("所有文件", "*.*"))
)
if not file_path:
messagebox.showinfo("提示", "未选择任何文件。")
return
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read().strip()
packets = content.split('\n\n')
output = []
for packet in packets:
lines = packet.strip().split('\n')
if len(lines) < 2:
continue
data_line = None
for line in lines:
if line.startswith('|'):
data_line = line
break
if not data_line:
continue
packet_bytes = hexstr_to_bytes(data_line)
if packet_bytes is None:
output.append(f"无法解析的数据行: {data_line[:50]}...")
continue
packet_result = process_packet_to_string(packet_bytes)
output.append(packet_result)
# 显示解析结果
text_area.delete(1.0, tk.END)
text_area.insert(tk.END, '\n\n'.join(output))
except Exception as e:
messagebox.showerror("错误", f"发生错误: {e}")
# 创建选择文件并分析的按钮
select_button = tk.Button(root, text="选择测试文件并分析", command=select_file)
select_button.pack(pady=10)
root.mainloop()
if __name__ == "__main__":
main_gui()
