import time from SimObjs import * from tkinter import * from tkinter import messagebox from PIL import ImageTk, Image from AgreementUtil import AgreementUtil def create_label(canvas: Canvas, x, y, width, height, text, rect_color, font_color, tag): canvas.create_rectangle(x, y, x + width, y + height, fill=rect_color, tags=tag) canvas.create_text(x + width / 2, y + height / 2, text=text, fill=font_color, tags=tag) return tag class PacketWindow(Toplevel): def __init__(self, master, packet, packet_step=["app", "trans", "ip", "mac"], *args, **kwargs): """ packet：{"packet": True, "mode": {"app": True, "trans": True, "ip": True, "mac": True}} """ super().__init__(master, *args, **kwargs) self.master = master self.width = 300 self.height = 400 self.geometry(f"{self.width}x{self.height}+200+200") self.attributes("-topmost", True) self.resizable(width=False, height=False) self.packet_button_str = {"app": "应用层", "ip": "网络层", "mac": "链路层", "trans": "传输层"} if not packet["packet"]: packet["mode"] = packet["mode"][::-1] # True 为封包 False 为解包 self.packet_step = packet_step self.step = 0 self.packet: dict[str: bool] = packet self.sendObj: SimHost = packet["sendObj"] self.receiveObj: SimHost = packet["receiveObj"] self.packet_option = {"app": {"color": "#ff3b26", "command": self.app_packet if packet["packet"] else self.app_unpack}, "ip": {"color": "#00a2f2", "command": self.ip_packet if packet["packet"] else self.ip_unpack}, "mac": {"color": "#46707a", "command": self.mac_packet if packet["packet"] else self.mac_unpack}, "trans": {"color": "#710a00", "command": self.trans_packet if packet["packet"] else self.trans_unpack}} self.canvas = Canvas(self, width=self.width, height=self.height) self.canvas.place(x=0, y=0, anchor='nw') self.packet_height, self.packet_init_width = 25, 100 self.background_img = ImageTk.PhotoImage(Image.open("../datas/images/背景@2x.png").resize((self.width, self.height))) self.canvas.create_image(0, 0, image=self.background_img, anchor=NW) # self.packet_rect = create_label(self.canvas, int(self.width / 2) - self.packet_init_width / 2, 10, width=self.packet_init_width, height=self.packet_height, text=self.packet["tag"], rect_color="#dee1e6", font_color="black", tag="packet") self.step_y = {} self.message = packet["tag"] self.create_widget() def move_to(self, tag, target_x, target_y, speed=10): # 获取当前位置 current_coords = self.canvas.coords(tag) if len(current_coords) < 2: return # 如果没有坐标，就退出函数 current_x, current_y = current_coords[0], current_coords[1] # 计算移动方向 delta_x = (target_x if target_x else current_x) - current_x delta_y = (target_y if target_y else current_y) - current_y # 计算下一步的位置 next_x = current_x + (speed if delta_x > 0 else -speed) if abs(delta_x) > speed else (target_x if target_x else current_x) next_y = current_y + (speed if delta_y > 0 else -speed) if abs(delta_y) > speed else (target_y if target_y else current_y) # 移动对象 self.canvas.move(tag, next_x - current_x, next_y - current_y) # 如果对象还没有到达目标，继续移动 if next_x != (target_x if target_x else current_x) or next_y != (target_y if target_y else current_y): self.canvas.after(10, lambda: self.move_to(tag, target_x, target_y, speed)) def create_widget(self): mode_text = "封包" if self.packet["packet"] else "解包" num, margin_top, button_width, button_height = 1, 30, 120, 40 for data in self.packet["mode"]: key = list(data.keys())[0] value = list(data.values())[0] if value: Button(self, text=self.packet_button_str[key] + mode_text, command=self.packet_option[key]["command"], font=("", 16)).place(x=40, y=30 + (margin_top + button_height) * (num - 1), width=button_width, height=button_height) num += 1 Button(self, text="发送", command=self.send_packet, font=("", 16)).place(x=self.width - 60, y=self.height - 40, anchor=NW) Button(self, text="取消", command=self.destroy, font=("", 16)).place(x=self.width - 120, y=self.height - 40, anchor=NW) def send_packet(self): if self.step != len(self.packet_step): messagebox.showerror("注意", "尚未完成{}！".format("封包" if self.packet["packet"] else "解包")) return self.packet["send"](self.message) self.destroy() def create_window(self, option): toplevel = Toplevel(self) toplevel.title(option["title"]) toplevel.geometry("450x220+300+300") for entry_option in option["entry"]: index = option["entry"].index(entry_option) key = list(entry_option.keys())[0] value = list(entry_option.values())[0] Label(toplevel, text=key, font=("", 16)).grid(row=index, column=0, padx=20, pady=10) if value is None: continue Entry(toplevel, textvariable=value, font=("", 16), width=15).grid(row=index, column=1, pady=10) Button(toplevel, text="提交", command=option["command"], font=("", 16)).place(x=450 - 60, y=220 - 40, anchor=NW) Button(toplevel, text="取消", command=toplevel.destroy, font=("", 16)).place(x=450 - 120, y=220 - 40, anchor=NW) toplevel.attributes("-topmost", True) return toplevel def app_packet(self): """ 应用层封包 """ if self.packet_step[self.step] != "app": messagebox.showerror("注意", "封包顺序出错！") return SourceAppAddr = StringVar() TargetAppAddr = StringVar() def packet(): toplevel.destroy() source_app_addr = SourceAppAddr.get() target_app_addr = TargetAppAddr.get() if source_app_addr == self.sendObj.interface[0]["addr"] \ and target_app_addr == self.receiveObj.interface[0]["addr"]: # 动画 # app_rect = create_label(self.canvas, self.width / 2 + 200, int(self.step_y["app"]), 50, 25, "AH", "#ff3b26", "white", tag="app") # self.move_to(self.packet_rect, None, int(self.step_y["app"]), speed=1) # self.move_to(app_rect, self.width / 2 + self.packet_init_width / 2, int(self.step_y["app"]), speed=1) self.message = target_app_addr + "&" + self.packet["tag"] self.master.message.show_message("应用层封包成功！数据包如下： \n{}".format(self.message)) self.step += 1 else: messagebox.showerror("提示", "应用层地址填写有误，请仔细检查！") toplevel = self.create_window({"title": "应用层封包", "entry": [{"发送主机应用层地址：": SourceAppAddr}, {"接收主机应用层地址：": TargetAppAddr}], "command": packet}) def app_unpack(self): """ 应用层解包 """ if self.packet_step[self.step] != "app": messagebox.showerror("注意", "解包顺序出错！") return TargetAppAddr = StringVar() def packet(): toplevel.destroy() target_app_addr = TargetAppAddr.get() if target_app_addr == self.receiveObj.interface[0]["addr"]: self.message = self.message.split("&")[1] self.master.message.show_message("应用层解包成功！数据包如下： \n{}".format(self.message)) self.step += 1 else: messagebox.showerror("提示", "应用层地址填写有误，请仔细检查！") toplevel = self.create_window({"title": "应用层解包", "entry": [{"接收主机应用层地址：": TargetAppAddr}], "command": packet}) def trans_packet(self): """ 传输层封包 """ if self.packet_step[self.step] != "trans": messagebox.showerror("注意", "封包顺序出错！") return SentPort = StringVar() RcptPort = StringVar() SplitCount = IntVar() def packet(): sent_port = SentPort.get() rcpt_port = RcptPort.get() split_count = SplitCount.get() count = split_appdata(self.packet["size"]) print(sent_port, self.sendObj.interface[0]["conn_port"]) if sent_port == self.sendObj.interface[0]["conn_port"] \ and rcpt_port == self.receiveObj.interface[0]["conn_port"]\ and split_count == count: toplevel.destroy() # 动画 # self.canvas.delete(self.packet_rect) # self.canvas.delete("app") # self.packet_rect = create_label(self.canvas, int(self.width / 2) - self.packet_init_width / 2, int(self.step_y["app"]), # width=self.packet_init_width + 50, height=self.packet_height, # text="A-" + self.packet["tag"], rect_color="#dee1e6", font_color="black", # tag="packet") # trans_rect = create_label(self.canvas, self.width / 2 + 200, int(self.step_y["trans"]), 50, 25, "PH", "#ff3b26", # "white", tag="trans") # self.move_to(self.packet_rect, None, int(self.step_y["trans"]), speed=1) # self.move_to(trans_rect, self.width / 2 + self.packet_init_width / 2 + 50, int(self.step_y["trans"]), speed=1) self.message = AgreementUtil.create_udp_packet(self.message, int(sent_port), int(rcpt_port)) self.master.message.show_message("传输层封包成功！数据包如下：\n{}".format(str([str(i + 1) + "-" + str(self.message) for i in range(split_count)]))) self.step += 1 else: messagebox.showerror("提示", "传输层封包信息填写有误，请仔细检查！") toplevel = self.create_window({"title": "传输层封包", "entry": [{"发送主机端口：": SentPort}, {"接收主机端口：": RcptPort}, {"拆包数量": SplitCount}, {"每个包约2048": None}], "command": packet}) def trans_unpack(self): """ 传输层解包 """ print(self.packet_step) print(self.step) if self.packet_step[self.step] != "trans": messagebox.showerror("注意", "解包顺序出错！") return RcptPort = StringVar() def packet(): rcpt_port = RcptPort.get() if rcpt_port == self.receiveObj.interface[0]["conn_port"]: toplevel.destroy() data = AgreementUtil.parse_udp_packet(self.message) print(data) self.message = data[-1] self.master.message.show_message("传输层解包成功！数据包如下：\n{}".format(self.message)) self.step += 1 else: messagebox.showerror("提示", "传输层解包信息填写有误，请仔细检查！") toplevel = self.create_window({"title": "传输层解包", "entry": [{"接收主机端口：": RcptPort}], "command": packet}) def ip_packet(self): """ 网络层封包 """ if self.packet_step[self.step] != "ip": messagebox.showerror("注意", "封包顺序出错！") return SourceIP = StringVar() TargetIP = StringVar() def packet(): source_ip = SourceIP.get() target_ip = TargetIP.get() if source_ip == self.sendObj.interface[0]["ip"] \ and target_ip == self.receiveObj.interface[0]["ip"]: toplevel.destroy() self.message = AgreementUtil.create_ip_packet(self.message, source_ip, target_ip) self.master.message.show_message("网络层封包成功！数据包如下：\n{}".format(self.message)) self.step += 1 else: messagebox.showerror("提示", "网络层封包信息填写有误，请仔细检查！") toplevel = self.create_window({"title": "网络层封包", "entry": [{"发送主机IP：": SourceIP}, {"接收主机IP：": TargetIP}], "command": packet}) def ip_unpack(self): """ 网络层解包 """ print(self.packet_step) print(self.step) if self.packet_step[self.step] != "ip": messagebox.showerror("注意", "解包顺序出错！") return RcptIP = StringVar() def packet(): rcpt_ip = RcptIP.get() if rcpt_ip == self.receiveObj.interface[0]["ip"]: toplevel.destroy() data = AgreementUtil.parse_ip_packet(self.message) print(data) self.message = data[-1] self.master.message.show_message("网络层解包成功！数据包如下：\n{}".format(self.message)) self.step += 1 else: messagebox.showerror("提示", "网络层解包信息填写有误，请仔细检查！") toplevel = self.create_window({"title": "网络层解包", "entry": [{"接收主机IP：": RcptIP}], "command": packet}) def mac_packet(self): """ 链路层封包 """ if self.packet_step[self.step] != "mac": messagebox.showerror("注意", "封包顺序出错！") return SentMac = StringVar() RcptMac = StringVar() def packet(): sent_mac = SentMac.get() rcpt_mac = RcptMac.get() if sent_mac == self.sendObj.interface[0]["mac"] \ and rcpt_mac == self.receiveObj.interface[0]["mac"]: toplevel.destroy() self.message = AgreementUtil.create_ethernet_frame(self.message, sent_mac, rcpt_mac) self.master.message.show_message("链路层封包成功！数据包如下：\n{}".format(self.message)) self.step += 1 else: messagebox.showerror("提示", "链路层封包信息填写有误，请仔细检查！") toplevel = self.create_window({"title": "链路层封包", "entry": [{"发送主机MAC：": SentMac}, {"接收主机MAC：": RcptMac}], "command": packet}) def mac_unpack(self): """ 链路层解包 """ if self.packet_step[self.step] != "mac": messagebox.showerror("注意", "解包顺序出错！") return RcptMac = StringVar() def packet(): rcpt_mac = RcptMac.get() if rcpt_mac == self.receiveObj.interface[0]["mac"]: toplevel.destroy() data = AgreementUtil.parse_ethernet_frame(self.packet["tag"]) print(data) self.message = data[-1] self.master.message.show_message("链路层解包成功！数据包如下：\n{}".format(self.message)) self.step += 1 else: messagebox.showerror("提示", "链路层解包信息填写有误，请仔细检查！") toplevel = self.create_window({"title": "链路层解包", "entry": [{"接收主机MAC：": RcptMac}], "command": packet}) class AppData: def __init__(self, obj_id, id_key, app_pack_id, app_pack_size, app_pack_tag, source_app_addr, target_app_addr, app_packed_string, timestamp, canvas): self.obj_id = obj_id self.id_key = id_key self.app_pack_id = app_pack_id self.app_pack_size = app_pack_size self.app_pack_tag = app_pack_tag self.source_app_addr = source_app_addr self.target_app_addr = target_app_addr self.app_packed_string = app_packed_string self.timestamp = timestamp def pack(self): # 为了简化，我们打包成一个字典 return vars(self) @staticmethod def unpack(packed_data): # 解包为AppData对象 return AppData(**packed_data) class TransData: def __init__(self, obj_id, id_key, trans_pack_id, trans_seq, trans_tag, app_pack_id, sent_port, rcpt_port, source_app_addr, target_app_addr, trans_packed_string, timestamp): self.obj_id = obj_id self.id_key = id_key self.trans_pack_id = trans_pack_id self.trans_seq = trans_seq self.trans_tag = trans_tag self.app_pack_id = app_pack_id self.sent_port = sent_port self.rcpt_port = rcpt_port self.source_app_addr = source_app_addr self.target_app_addr = target_app_addr self.trans_packed_string = trans_packed_string self.timestamp = timestamp def pack(self): return vars(self) @staticmethod def unpack(packed_data): return TransData(**packed_data) class IPData: def __init__(self, obj_id, id_key, ip_pack_id, trans_pack_id, source_ip, target_ip, source_app_addr, target_app_addr, ip_packed_string): self.ObjID = obj_id self.IDkey = id_key self.IPPackID = ip_pack_id self.TransPackID = trans_pack_id self.SourceIP = source_ip self.TargetIP = target_ip self.SourceAppAddr = source_app_addr self.TargetAppAddr = target_app_addr self.IPPackedString = ip_packed_string def pack(self): return vars(self) @staticmethod def unpack(packed_data): return IPData(**packed_data) class MACData: def __init__(self, obj_id, id_key, mac_pack_id, ip_pack_id, sent_mac, rcpt_mac, source_ip, target_ip, source_app_addr, target_app_addr, mac_packed_string): self.ObjID = obj_id self.IDkey = id_key self.MacPackID = mac_pack_id self.IPPackID = ip_pack_id self.SentMAC = sent_mac self.RcptMAC = rcpt_mac self.SourceIP = source_ip self.TargetIP = target_ip self.SourceAppAddr = source_app_addr self.TargetAppAddr = target_app_addr self.MacPackedString = mac_packed_string def pack(self): return vars(self) @staticmethod def unpack(packed_data): return MACData(**packed_data) def split_appdata(AppPackSize): MTU = 2048 return AppPackSize // MTU + (AppPackSize % MTU > 0) if __name__ == '__main__': # 假设的最大传输单元（MTU） MTU = 2048 # bytes # 模拟的数据和数据大小 SIMULATED_DATA = "hello" SIMULATED_SIZE = 2049 # 10 MB # 创建应用层数据包 app_packet = AppData("123", "123", "app1", SIMULATED_SIZE, "DATA", "192.0.2.1", "198.51.100.1", SIMULATED_DATA, time.time()) packed_app_data = app_packet.pack() print(packed_app_data)