NetWork_v1/packet.py

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)