Python网络编程实现DHCP服务器

import tkinter as tk from tkinter import ttk import threading import socket import struct class DHCP_Server_GUI: def __init__(self, root): self.root = root self.root.title("DHCP 服务器") self.root.geometry("400x300") self.start_button = ttk.Button(self.root, text="启动服务器", command=self.start_server) self.start_button.pack(pady=20) self.stop_button = ttk.Button(self.root, text="停止服务器", command=self.stop_server, state=tk.DISABLED) self.stop_button.pack(pady=10) self.status_label = ttk.Label(self.root, text="服务器未启动", font=("Arial", 12)) self.status_label.pack(pady=10) self.client_info_treeview = ttk.Treeview(self.root, columns=("IP", "MAC"), show="headings") self.client_info_treeview.heading("IP", text="IP 地址") self.client_info_treeview.heading("MAC", text="MAC 地址") self.client_info_treeview.pack(pady=10) self.server_socket = None self.server_thread = None self.stop_event = threading.Event() def start_server(self): self.start_button.config(state=tk.DISABLED) self.stop_button.config(state=tk.NORMAL) self.status_label.config(text="服务器运行中", foreground="green") self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) self.server_socket.bind(('192.168.1.5', 67)) # self.server_socket.bind(('0.0.0.0', 67)) self.server_thread = threading.Thread(target=self.run_server) self.server_thread.start() def stop_server(self): self.start_button.config(state=tk.NORMAL) self.stop_button.config(state=tk.DISABLED) self.status_label.config(text="服务器已停止", foreground="red") self.stop_event.set() self.server_thread.join() self.server_socket.close() def run_server(self): while not self.stop_event.is_set(): try: data, addr = self.server_socket.recvfrom(1024) print("data===",data,len(data)) print("addr===",addr) request_packet = self.parse_dhcp_packet(data) print("request_packet===",request_packet,len(request_packet)) # 判断是否是DHCP报文 if request_packet[0] == 1: # 创建DHCP Offer报文 offer_packet = self.create_dhcp_offer(request_packet) # decimal_value = struct.unpack('!BB', request_packet[11:12]) print("offer_packet===",offer_packet,len(offer_packet)) self.server_socket.sendto(offer_packet, addr) self.update_client_info(addr[0], request_packet[11:12]) except socket.timeout: continue def parse_dhcp_packet(self, data): dhcp_format = '!BBBBLHHLLLL16s64s128sBBBBBB' # dhcp_format = '!1s1s1s1s4s2s2s4s4s4s4s2B8B16B4s' print("dhcp_format==",dhcp_format,len(dhcp_format)) # 你需要确保提供给 struct.unpack 函数的数据长度与格式字符串所需的长度一致。 # 你还需要确保格式字符串 dhcp_format 的正确性。请确保该格式字符串与你要解包的数据的结构相匹配，包括字段的数量、类型和顺序。 request_packet = struct.unpack(dhcp_format, data) # print("request_packet--",len(request_packet),request_packet) return request_packet def create_dhcp_offer(self,discover_packet): discover_packet = { 'xid': b'\x01\x02\x03\x04', 'chaddr': b'\x00\x11\x22\x33\x44\x55\x66', } # 构建 DHCP Offer 报文字段值 # mac_address = '00:11:22:33:44:55' # mac_address_bytes = bytes.fromhex(mac_address.split(':','')) # mac_address_bytes = bytes.fromhex(''.join(mac_address.split(':'))) mac_address_bytes = discover_packet['chaddr'] print("mac_address_bytes==",mac_address_bytes) print("(mac_address_bytes') ===",int.from_bytes(mac_address_bytes, 'big')) print("int.from_bytes(discover_packet['xid'], 'big') ===",int.from_bytes(discover_packet['xid'], 'big')) print("*map(int, socket.inet_aton('192.168.1.5'))==",*map(int, socket.inet_aton('192.168.1.5'))) offer_packet = struct.pack( # '!BBBB I 4s HH 4s4s4s4s 16s 64s 128s', '!BBBBIHH4s4s 4s4s 16s 64s 128sI', # '!BBBBI4sHH4 s4s4s4s16s64s128sI', 2, # OP: 2 (Boot Reply) 1, # HTYPE: 1 (Ethernet) 6, # HLEN: 6 (MAC Address Length) 0, # HOPS: 0 (Hops) I int.from_bytes(discover_packet['xid'], 'big'), # XID: 4 bytes (Transaction ID) # 4s 0, # SECS: 2 bytes (Seconds) 0, # FLAGS: 2 bytes (Flags) b'\x00\x00\x00\x00', # CIADDR: 4 bytes (Client IP Address) b'\x00\x00\x00\x00', # YIADDR: 4 bytes (Your IP Address) # # HH b'\x00\x00\x00\x00', # SIADDR: 4 bytes (Server IP Address) b'\x00\x00\x00\x00', # GIADDR: 4 bytes (Gateway IP Address) # # 16s mac_address_bytes, # CHADDR: 16 bytes (Client MAC Address) b'\x00' * 64, # SNAME: 64 bytes (Server Name) b'\x00' * 128, # FILE: 128 bytes (Boot File Name) 0x63825363, # OPT: 4 bytes (Magic Cookie) ) offer_packet += struct.pack( '!BBB', # DHCP Option: 53 (DHCP Message Type) 53, # Option Code 1, # Option Length 2, # Option Value (Offer) ) offer_packet += struct.pack( '!BBBBBB', # DHCP Option: 54 (DHCP Server Identifier) 54, # Option Code 4, # Option Length *map(int, socket.inet_aton('192.168.1.5')), # Option Value (Server IP Address) ) offer_packet += struct.pack( '!BBBBBB', # DHCP Option: 50 (Requested IP Address) 50, # Option Code 4, # Option Length *map(int, socket.inet_aton('192.168.1.6')), # Option Value (Client Requested IP Address) ) offer_packet += struct.pack( '!B', # DHCP Option: 255 (End of Options) 255, # Option Code ) # 输出 DHCP Offer 报文字段值 print(offer_packet) return offer_packet def update_client_info(self, ip, mac): self.client_info_treeview.insert("", tk.END, values=(ip, mac)) if __name__ == "__main__": root = tk.Tk() dhcp_server_gui = DHCP_Server_GUI(root) root.mainloop()