Python-web-communicate/main.py

import tkinter as tk
from tkinter import ttk, messagebox
import socket
import struct
import time
import uuid
import re

# 获取本机信息
def get_local_info():
    """
    获取本机的主机名、MAC地址和IP地址列表。

    Returns:
        tuple: 包含主机名、MAC地址和IP地址列表的元组。
    """
    hostname = socket.gethostname()  # 获取主机名
    local_ips = socket.gethostbyname_ex(hostname)[-1]  # 获取本机的IP地址列表
    mac = ':'.join(re.findall('..', '%012x' % uuid.getnode()))  # 获取MAC地址
    return hostname, mac, local_ips

# 验证 IP 地址或域名
def validate_ip_or_domain(value):
    """
    验证输入值是否为有效的IP地址或域名。

    Args:
        value (str): 待验证的IP地址或域名。

    Returns:
        bool: 如果输入值有效返回True，否则返回False。
    """
    ip_pattern = re.compile(r'^\d{1,3}(\.\d{1,3}){3}$')  # 匹配IPv4地址的正则
    domain_pattern = re.compile(r'^([a-zA-Z0-9-]+\.)+[a-zA-Z]{2,}$')  # 匹配域名的正则
    return ip_pattern.match(value) or domain_pattern.match(value)

# 验证端口号
def validate_port(port):
    """
    验证输入的端口号是否有效。

    Args:
        port (str): 待验证的端口号。

    Returns:
        bool: 如果端口号有效返回True，否则返回False。
    """
    return port.isdigit() and 1 <= int(port) <= 65535

# 构造 IP 头部
def construct_ip_header(src_ip, dest_ip, payload_length, protocol):
    """
    构造IP头部。

    Args:
        src_ip (str): 源IP地址。
        dest_ip (str): 目标IP地址。
        payload_length (int): 负载长度。
        protocol (int): 协议号。

    Returns:
        bytes: 构造的IP头部。
    """
    version = 4
    ihl = 5
    tos = 0
    total_length = 20 + payload_length  # IP头部20字节 + 数据部分
    identification = 54321
    flags_offset = 0
    ttl = 64
    header_checksum = 0

    src_ip_bytes = socket.inet_aton(src_ip)  # 将源IP转换为二进制格式
    dest_ip_bytes = socket.inet_aton(dest_ip)  # 将目标IP转换为二进制格式
    ip_header = struct.pack('!BBHHHBBH4s4s',
                            (version << 4) + ihl, tos, total_length,
                            identification, flags_offset, ttl, protocol,
                            header_checksum, src_ip_bytes, dest_ip_bytes)

    return ip_header

# 构造 ICMP 报文
def construct_icmp():
    """
    构造ICMP报文。

    Returns:
        bytes: 构造的ICMP报文。
    """
    # ICMP 报文头部: 类型(8), 代码(0), 校验和(0), 标识符(1), 序列号(1)
    icmp_header = struct.pack('!BBHHH', 8, 0, 0, 1, 1)
    data = b'Ping'
    # 计算校验和
    checksum = calculate_checksum(icmp_header + data)
    # 填充校验和字段
    icmp_header = struct.pack('!BBHHH', 8, 0, checksum, 1, 1)
    return icmp_header + data

# 构造 UDP 报文
def construct_udp(src_port, dest_port):
    """
    构造UDP报文。

    Args:
        src_port (int): 源端口号。
        dest_port (int): 目标端口号。

    Returns:
        bytes: 构造的UDP报文。
    """
    # UDP头部格式: 源端口号 (2 字节), 目标端口号 (2 字节), 长度 (2 字节), 校验和 (2 字节)
    udp_header = struct.pack('!HHHH', src_port, dest_port, 8, 0)  # 长度固定为8，校验和暂时为0
    return udp_header

# 构造 TCP 报文（SYN）
def construct_tcp(src_port, dest_port):
    """
    构造TCP报文（SYN）。

    Args:
        src_port (int): 源端口号。
        dest_port (int): 目标端口号。

    Returns:
        bytes: 构造的TCP报文。
    """
    # 初始化TCP标头字段
    seq = 0  # 初始序列号
    ack_seq = 0  # 确认序列号
    offset = 5  # TCP头部的长度，单位是32位字（5 * 4 = 20字节）
    reserved = 0  # 保留字段，默认为0
    flags = 0b000010  # SYN标志位设置为1，其他标志为0
    window = socket.htons(5840)  # 窗口大小，使用network字节序
    checksum = 0  # 校验和，暂时为0
    urgent_ptr = 0  # 紧急指针，默认为0

    # 打包TCP报文头部
    tcp_header = struct.pack('!HHLLBBHHH',
                             src_port, dest_port, seq, ack_seq,
                             (offset << 4) + reserved, flags, window,
                             checksum, urgent_ptr)
    return tcp_header

# 计算校验和
def calculate_checksum(data):
    """
    计算校验和。

    Args:
        data (bytes): 待计算校验和的数据。

    Returns:
        int: 计算得到的校验和。
    """
    checksum = 0
    n = len(data)

    # 按 16 位块划分
    for i in range(0, n - 1, 2):
        chunk = (data[i] << 8) + data[i + 1]
        checksum += chunk

    # 如果长度为奇数，补零
    if n % 2 == 1:
        checksum += data[-1] << 8

    # 将 32 位总和折叠为 16 位
    checksum = (checksum >> 16) + (checksum & 0xFFFF)
    checksum += (checksum >> 16)

    # 对总和取反
    return ~checksum & 0xFFFF

# 发送报文
def send_packet():
    """
    发送报文。
    """
    try:
        src_ip = ip_combobox.get()
        dest_host = dest_entry.get()
        dest_port = port_entry.get()
        packet_type = var.get()

        # 验证目标地址
        if not validate_ip_or_domain(dest_host):
            raise ValueError("目标地址无效，必须为有效的 IP 或域名")

        # 验证端口号
        if not validate_port(dest_port):
            raise ValueError("端口号无效，必须是有效的正整数（1-65535）")

        dest_ip = socket.gethostbyname(dest_host)
        src_port = 12345
        dest_port = int(dest_port)

        # ICMP报文需要添加 RTT 计算
        if packet_type == "ICMP":
            sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP)
            packet = construct_icmp()

            # 记录发送时间
            start_time = time.time()

            # 发送ICMP报文
            sock.sendto(packet, (dest_ip, 0))

            # 接收响应
            sock.settimeout(2)  # 设置超时为2秒
            try:
                sock.recvfrom(1024)
                end_time = time.time()
                rtt = (end_time - start_time) * 1000  # 毫秒
                result.set(f"ICMP请求成功，RTT: {rtt:.2f} 毫秒")
            except socket.timeout:
                result.set("ICMP请求超时")
            finally:
                sock.close()

            return  # 结束函数

        # 对于其他报文类型（UDP, TCP）
        if packet_type == "UDP":
            sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            packet = construct_udp(src_port, dest_port)
        elif packet_type == "TCP":
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.connect((dest_ip, dest_port))
            packet = b"Hello, TCP!"  # TCP 是面向连接的，这里发送简单的字符串
            sock.sendall(packet)
            sock.close()
            result.set(f"TCP 连接成功，数据已发送！")
            return  # 结束函数

        # 发送报文
        sock.sendto(packet, (dest_ip, dest_port))
        result.set(f"报文发送成功！类型: {packet_type}")
    except socket.gaierror:
        messagebox.showerror("网络错误", "无法解析目标地址，请检查输入的 IP 或域名是否正确。")
    except socket.error as e:
        messagebox.showerror("网络错误", f"网络错误: {e.strerror}")
    except ValueError as e:
        messagebox.showerror("输入错误", f"输入无效: {e}")
    except Exception as e:
        messagebox.showerror("未知错误", f"发送报文时发生未知错误: {e}")

# GUI 界面设计
window = tk.Tk()
window.title("网络通信软件")

# 获取本机信息
hostname, mac, local_ips = get_local_info()

tk.Label(window, text="本机信息:").grid(row=0, column=0)
info_text = tk.Text(window, height=4, width=40)
info_text.grid(row=0, column=1, columnspan=2)
info_text.insert(tk.END, f"主机名: {hostname}\nMAC 地址: {mac}\n可用 IP: {', '.join(local_ips)}")
info_text.config(state=tk.DISABLED)

# 配置IP地址选择
tk.Label(window, text="本机 IP 地址:").grid(row=1, column=0)
ip_combobox = ttk.Combobox(window, values=local_ips)
ip_combobox.grid(row=1, column=1)
ip_combobox.current(0)

# 配置目标地址输入框
tk.Label(window, text="目标地址:").grid(row=2, column=0)
dest_entry = tk.Entry(window)
dest_entry.grid(row=2, column=1)

# 配置目标端口输入框
tk.Label(window, text="目标端口:").grid(row=3, column=0)
port_entry = tk.Entry(window)
port_entry.grid(row=3, column=1)

# 配置报文类型选择
tk.Label(window, text="报文类型:").grid(row=4, column=0)
var = tk.StringVar(value="ICMP")
tk.Radiobutton(window, text="ICMP", variable=var, value="ICMP").grid(row=4, column=1)
tk.Radiobutton(window, text="UDP", variable=var, value="UDP").grid(row=4, column=2)
tk.Radiobutton(window, text="TCP", variable=var, value="TCP").grid(row=4, column=3)

# 配置结果显示
result = tk.StringVar()
tk.Label(window, textvariable=result).grid(row=6, column=0, columnspan=3)

# 发送报文按钮
tk.Button(window, text="发送报文", command=send_packet).grid(row=5, column=1)

# 启动GUI主循环
window.mainloop()