说明
本系统模拟实现了一个路由器与两个主机节点。该路由器将接收原始以太网帧，并像真正的路由器一样处理它们：将它们转发到正确的传出接口，处理以太网帧，处理 IPv4 分组，处理 ARP分组，处理 ICMP 分组，创建新帧等。这个路由器将模拟实现路由器处理主机节点发送的以太网帧的操作。
运行展示
源代码
import queue
import threading
import time
import socket
import uuid
# 十进制转二进制
def dec_to_bin_str(num):
return str(bin(num))[2:]
# 二进制转十进制
def bin_to_dec_str(binary):
return str(int(binary, 2))
# IP地址转二进制
def ipv4_to_binary(ipv4):
binary_list = ['{0:08b}'.format(int(num)) for num in ipv4.split('.')]
return ''.join(binary_list)
# 二进制地址转点分十进制
def binary_to_ipv4(binary_str):
if len(binary_str) != 32:
return "Invalid binary string"
segments = [binary_str[i:i+8] for i in range(0, 32, 8)]
ipv4_address = ".".join(str(int(segment, 2)) for segment in segments)
return ipv4_address
# 验证校验和
def validate_ip_checksum(data):
# 初始化总合为0
total = 0
# 每16位为单位进行遍历
for i in range(0, len(data), 16):
word = int(data[i:i + 16], 2)
total += word
# 按位与操作和右移操作，在总和的低16位中和高16位中分别加上结果
total = (total & 0xffff) + (total >> 16)
# 再次按位与操作和右移操作，对结果再次进行相加
total = (total + (total >> 16)) & 0xffff
# 将总和与0xffff进行异或运算，并将结果转换为16位二进制数
checksum = '{:016b}'.format(total ^ 0xffff)
return checksum
# IP数据报的格式（IPv4格式）
class IPv4Message:
def __init__(self, version, header_length, service, total_length, identification, flags, fragment_offset, ttl, protocol, checksum, source_address, destination_address, data):
self.version = version # 版本
self.header_length = header_length # 首部长度
self.service = service # 区分服务
self.total_length = total_length # 总长度
self.identification = identification # 标识
self.flags = flags # 标志
self.fragment_offset = fragment_offset # 片偏移
self.ttl = ttl # 生存时间
self.protocol = protocol # 协议
self.checksum = checksum # 校验和
self.source_address = source_address # 源地址
self.destination_address = destination_address # 目的地址
self.data = data # 数据
# 以太网MAC帧的格式
class MACMessage:
def __init__(self, mac_destination, mac_source, protocol_type, data):
self.mac_destination = mac_destination
self.mac_source = mac_source
self.protocol_type = protocol_type # 为IPv4
self.data = data
# ICMP差错报文格式
class ICMPError:
def __init__(self, message_type, checksum, data):
self.type = message_type # 1-主机不可达
self.checksum = checksum
self.data = data
# ICMP回送请求或回答报文格式
class ICMPMessage:
def __init__(self, message_type, checksum, data):
self.type = message_type # 类型字段，8-请求，0-回答
self.checksum = checksum
self.data = data
# ARP广播格式
class ARPBroadcast:
def __init__(self):
self.mac = "FF:FF:FF:FF:FF:FF"
# 获取本机IP与本机MAC
my_ip = socket.gethostbyname(socket.gethostname())
my_mac = ':'.join(['{:02x}'.format((uuid.getnode() >> elements) & 0xff) for elements in range(0, 2 * 6, 2)][::-1])
# ARP表
arp_table = {my_ip: my_mac, '172.26.213.121': '00:11:22:33:44:55', '172.26.213.64': '00:aa:bb:cc:dd:ee'}
# 路由表
route_table = {my_ip: 'direct', '172.26.213.121': 'direct', '0.0.0.0': '172.26.213.64'}
# 路由器类
class Router:
def __init__(self):
self.ip = my_ip
self.mac = my_mac
self.arp_table = arp_table
self.routing_table = route_table
self.frame_queue = queue.Queue()
def route(self):
while True:
# 拆开frame获取IP数据报
frame_message = self.frame_queue.get()
if frame_message == "":
continue
ip_message = frame_message.split("@")[-1]
ip_header = ip_message[0:160]
# 计算校验和
checksum = ip_header[80:96]
rest = ip_header[-64:]
print("路由器：计算的校验和为：" + checksum)
ip_message_t = ip_header[0:80]
ip_message_t += "0000000000000000"
ip_message_t += rest
if not checksum == validate_ip_checksum(ip_message_t):
print("路由器：校验和验证失败")
# 发送一个ICMP响应报文
print("路由器：发送一个ICMP响应报文告知主机")
continue
print("路由器：校验和验证成功")
# 验证TTL合法性
ttl = int(bin_to_dec_str(ip_header[64:72]))
if ttl <= 0:
print("TTL值不合法")
# 发送一个ICMP响应报文
print("路由器：发送一个ICMP响应报文告知主机")
continue
else:
# 自减
ttl -= 1
# 拆解IP首部
ip_source = binary_to_ipv4((ip_message[0:160])[-64:-32])
ip_destination = binary_to_ipv4((ip_message[0:160])[-32:])
print("路由器：源IP：" + ip_source + "，目的IP：" + ip_destination)
# 转发的frame
new_frame_message = ""
# 根据路由表查目的IP地址
for ip, move in route_table.items():
if ip == ip_destination:
# 修改IP头部，源地址改变但目的地址不变
if move == "direct":
print("路由器：目标可直达，直接转发")
elif move == "upper":
print("路由器：目标为本身，交付上层")
break
else:
ip_header = ip_header[0:-32]
ip_header += ipv4_to_binary(route_table["0.0.0.0"])
print("路由器：新的IP数据报已生成，下一跳IP地址为：" + route_table["0.0.0.0"])
# 根据ARP表获取下一跳MAC，封装成帧，转发
mac_source = self.mac
for m_ip, mac in arp_table.items():
if ip_header[-32:] == ipv4_to_binary(m_ip):
mac_destination = mac
new_frame_message += mac_destination + "@" + mac_source + "@IPv4@" + ip_message
print("路由器：路由器已转发，下一跳MAC为：", mac)
# 主机节点类
class Host:
def __init__(self, ip, mac):
self.ip = ip
self.mac = mac
# 获取目的MAC地址(广播ARP请求分组并接收ARP响应分组)
def get_dest_mac(self, dest_ip):
print("主机：正在进行ARP广播")
dest_mac = "FF:FF:FF:FF:FF:FF"
for ip, mac in arp_table.items():
if ip == dest_ip:
dest_mac = my_mac
print("主机：目标MAC已获取" + dest_mac)
return dest_mac
# 初始化校验和
def make_checksum(self, ip_header, dest_ip):
ip_header += "0000000000000000"
ip_header += ipv4_to_binary(self.ip) + ipv4_to_binary(dest_ip)
data = ip_header
total = 0
for i in range(0, len(data), 16):
word = int(data[i:i + 16], 2)
total += word
total = (total & 0xffff) + (total >> 16)
total = (total + (total >> 16)) & 0xffff
checksum = '{:016b}'.format(total ^ 0xffff)
print("主机：生成的校验和为：" + checksum)
return checksum
# 发送数据帧
def send_frame(self, dest_ip, data):
# 组装IP数据报
ip_header = "0100" # 版本号为4
ip_header += "0101" # 首部长度20B
ip_header += "00000000" # 区分服务
ip_header += dec_to_bin_str(len(data) + 20).zfill(16) # 总长度
ip_header += "0000000000000000" # 标识
ip_header += "000" # 标志
ip_header += "0000000000000" # 片偏移
ip_header += "00000000" # 生存时间
ip_header += "00000000" # 协议
ip_header += self.make_checksum(ip_header, dest_ip) # 校验和
ip_header += ipv4_to_binary(self.ip) # 源地址
ip_header += ipv4_to_binary(dest_ip) # 目的地址
ip_message = ip_header + data # 组装成IP数据报
# 组装数据帧
frame_head = self.get_dest_mac(dest_ip)
frame_head += "@" + self.mac + "@IPv4@"
frame_message = frame_head + ip_message
# 发送给路由器
print("主机：数据帧发送完毕")
my_router.frame_queue.put(frame_message)
# 路由器与主机节点
my_router = Router()
my_host1 = Host("172.26.213.121", "00:11:22:33:44:55")
my_host2 = Host("172.26.213.122", "00:11:22:33:44:55")
# 打开线程
router_thread = threading.Thread(target=my_router.route)
router_thread.start()
# 标志位
flag = True
# 打印本机IP与MAC，即路由器IP、MAC
print("本机IP：" + my_ip + " 本机MAC：" + my_mac)
# 轮询输入
while True:
message = input("主机：请输入要发送的消息：")
if message == "exit":
print("主机已关闭")
break
if message == "shift":
flag = not flag
continue
if flag:
my_host1.send_frame(my_ip, message)
else:
my_host2.send_frame("172.26.21.12", message)
time.sleep(1)
# 释放资源，关闭线程
router_thread.join()
文章链接
发表评论