feat: display self info +tcp +udp
This commit is contained in:
parent
2559eaf66c
commit
300f42fd8f
172
main.py
172
main.py
@ -1,102 +1,172 @@
|
||||
import tkinter as tk
|
||||
from tkinter import messagebox
|
||||
from tkinter import ttk, messagebox
|
||||
import socket
|
||||
import struct
|
||||
import re
|
||||
import time
|
||||
import uuid
|
||||
import re
|
||||
|
||||
# 获取本机信息
|
||||
def get_local_info():
|
||||
hostname = socket.gethostname()
|
||||
local_ips = socket.gethostbyname_ex(hostname)[-1]
|
||||
mac = ':'.join(re.findall('..', '%012x' % uuid.getnode()))
|
||||
return hostname, mac, local_ips
|
||||
|
||||
# 计算校验和
|
||||
def calculate_checksum(data):
|
||||
checksum = 0
|
||||
n = len(data)
|
||||
# 验证 IP 地址或域名
|
||||
def validate_ip_or_domain(value):
|
||||
ip_pattern = re.compile(r'^\d{1,3}(\.\d{1,3}){3}$')
|
||||
domain_pattern = re.compile(r'^([a-zA-Z0-9-]+\.)+[a-zA-Z]{2,}$')
|
||||
return ip_pattern.match(value) or domain_pattern.match(value)
|
||||
|
||||
# 按 16 位块划分
|
||||
for i in range(0, n - 1, 2):
|
||||
chunk = (data[i] << 8) + data[i + 1]
|
||||
checksum += chunk
|
||||
# 验证端口号
|
||||
def validate_port(port):
|
||||
return port.isdigit() and 1 <= int(port) <= 65535
|
||||
|
||||
# 如果长度为奇数,补零
|
||||
if n % 2 == 1:
|
||||
checksum += data[-1] << 8
|
||||
# 构造 IP 头部
|
||||
def construct_ip_header(src_ip, dest_ip, payload_length, protocol):
|
||||
version = 4
|
||||
ihl = 5
|
||||
tos = 0
|
||||
total_length = 20 + payload_length
|
||||
identification = 54321
|
||||
flags_offset = 0
|
||||
ttl = 64
|
||||
header_checksum = 0
|
||||
|
||||
# 将 32 位总和折叠为 16 位
|
||||
checksum = (checksum >> 16) + (checksum & 0xFFFF)
|
||||
checksum += (checksum >> 16)
|
||||
|
||||
# 对总和取反
|
||||
return ~checksum & 0xFFFF
|
||||
src_ip_bytes = socket.inet_aton(src_ip)
|
||||
dest_ip_bytes = socket.inet_aton(dest_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 报文头部: 类型(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_header = struct.pack('!HHHH', src_port, dest_port, 8, 0)
|
||||
return udp_header
|
||||
|
||||
# 构造 TCP 报文(SYN)
|
||||
def construct_tcp(src_port, dest_port):
|
||||
seq = 0
|
||||
ack_seq = 0
|
||||
offset = 5
|
||||
reserved = 0
|
||||
flags = 0b000010 # SYN flag
|
||||
window = socket.htons(5840)
|
||||
checksum = 0
|
||||
urgent_ptr = 0
|
||||
|
||||
# 验证输入
|
||||
def validate_ip(ip):
|
||||
pattern = re.compile(r'^\d{1,3}(\.\d{1,3}){3}$')
|
||||
return pattern.match(ip) is not None
|
||||
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):
|
||||
checksum = 0
|
||||
n = len(data)
|
||||
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
|
||||
checksum = (checksum >> 16) + (checksum & 0xFFFF)
|
||||
checksum += (checksum >> 16)
|
||||
return ~checksum & 0xFFFF
|
||||
|
||||
# 发送报文
|
||||
def send_packet():
|
||||
try:
|
||||
target_ip = ip_entry.get()
|
||||
target_port = int(port_entry.get())
|
||||
src_ip = ip_combobox.get()
|
||||
dest_host = dest_entry.get()
|
||||
dest_port = port_entry.get()
|
||||
packet_type = var.get()
|
||||
if not validate_ip(target_ip):
|
||||
raise ValueError("目标 IP 地址无效")
|
||||
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_RAW,
|
||||
socket.IPPROTO_ICMP if packet_type == "ICMP" else socket.IPPROTO_UDP)
|
||||
# 验证目标地址
|
||||
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)
|
||||
|
||||
# 构造报文并发送
|
||||
if packet_type == "ICMP":
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP)
|
||||
packet = construct_icmp()
|
||||
elif packet_type == "UDP":
|
||||
packet = construct_udp(12345, target_port)
|
||||
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 # 结束函数
|
||||
|
||||
start_time = time.time()
|
||||
sock.sendto(packet, (target_ip, target_port))
|
||||
end_time = time.time()
|
||||
|
||||
result.set(f"报文发送成功!类型: {packet_type}, 往返时间: {end_time - start_time:.2f} 秒")
|
||||
# 发送报文
|
||||
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}")
|
||||
|
||||
messagebox.showerror("未知错误", f"发送报文时发生未知错误: {e}")
|
||||
|
||||
# GUI 界面设计
|
||||
window = tk.Tk()
|
||||
window.title("网络通信软件")
|
||||
|
||||
tk.Label(window, text="目标 IP 地址:").grid(row=0, column=0)
|
||||
ip_entry = tk.Entry(window)
|
||||
ip_entry.grid(row=0, column=1)
|
||||
hostname, mac, local_ips = get_local_info()
|
||||
|
||||
tk.Label(window, text="目标端口号:").grid(row=1, column=0)
|
||||
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)
|
||||
|
||||
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=1, column=1)
|
||||
port_entry.grid(row=3, column=1)
|
||||
|
||||
tk.Label(window, text="报文类型:").grid(row=2, column=0)
|
||||
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=2, column=1)
|
||||
tk.Radiobutton(window, text="UDP", variable=var, value="UDP").grid(row=2, column=2)
|
||||
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=4, column=0, columnspan=3)
|
||||
tk.Label(window, textvariable=result).grid(row=6, column=0, columnspan=3)
|
||||
|
||||
tk.Button(window, text="发送报文", command=send_packet).grid(row=3, column=1)
|
||||
tk.Button(window, text="发送报文", command=send_packet).grid(row=5, column=1)
|
||||
window.mainloop()
|
||||
|
Loading…
Reference in New Issue
Block a user