实验三 Socket编程技术实验报告
姓名：刘人宁
学号：20253201
队友：庄景博（学号：20253214）
实验环境：Windows 物理机
开发工具：PyCharm、trea
实验日期：2026.04.27
一、实验目的
利用 Python 编写 TCP 服务端与客户端，基于 Socket 完成点对点通信。
熟练掌握 IP+端口绑定、监听、连接、数据收发完整流程。
了解加密通信机制与数据安全传输原理，具备密码通信拓展能力。
使用 LLM 生成创意单人本地GUI程序，学习图形界面开发。
交替充当客户端、服务端，与队友完成双向通信，代码上传码云托管。
二、实验原理
TCP 是面向连接的可靠传输协议，通信前需三次握手建立连接。
服务端：创建套接字→绑定地址端口→监听→等待客户端接入
客户端：创建套接字→主动连接服务端 IP 与端口
双方建立连接后，通过 send()、recv() 完成数据双向传输
本次联调使用明文传输便于调试，已掌握对称加密、加解密传输理论与实现方法，可快速改造为密文通信。
三、基础控制台通信实现
3.1
服务端代码（SockeServer.py）
`import tkinter as tk
from tkinter import scrolledtext, messagebox, filedialog
import socket
import threading
import os
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
import base64
KEY = b"1234567890123456"
IV = b"abcdefghijklmnop"
def encrypt(msg):
cipher = AES.new(KEY, AES.MODE_CBC, IV)
encrypted = cipher.encrypt(pad(msg.encode('utf-8'), AES.block_size))
return base64.b64encode(encrypted).decode('utf-8')
def decrypt(msg):
cipher = AES.new(KEY, AES.MODE_CBC, IV)
decrypted = unpad(cipher.decrypt(base64.b64decode(msg)), AES.block_size)
return decrypted.decode('utf-8')
def send_file(conn, file_path):
if not os.path.exists(file_path):
return "文件不存在"
file_name = os.path.basename(file_path)
file_size = os.path.getsize(file_path)conn.send(f"FILE:{file_name}:{file_size}".encode("utf-8"))confirm = conn.recv(1024).decode("utf-8")
if confirm != "READY":return "传输被拒绝"with open(file_path, "rb") as f:while True:data = f.read(1024)if not data:breakconn.send(data)
return f"文件 {file_name} 发送完成"
def receive_file(conn):
file_info = conn.recv(1024).decode("utf-8")
if not file_info.startswith("FILE:"):
return None
parts = file_info.split(":")
if len(parts) < 3:return Nonefile_name = parts[1]
file_size = int(parts[2])conn.send("READY".encode("utf-8"))received_size = 0
with open(file_name, "wb") as f:while received_size < file_size:data = conn.recv(1024)if not data:breakf.write(data)received_size += len(data)return file_name
class ServerGUI:
def init(self, root):
self.root = root
self.root.title("加密聊天服务器")
self.root.geometry("600x500")
self.conn = Noneself.running = Falseself.create_widgets()def create_widgets(self):self.status_label = tk.Label(self.root, text="状态: 未连接", fg="red")self.status_label.pack(pady=5)self.info_label = tk.Label(self.root, text="服务器地址: 192.168.123.1:8888")self.info_label.pack()self.start_button = tk.Button(self.root, text="启动服务器", command=self.start_server, bg="lightblue")self.start_button.pack(pady=5)self.close_button = tk.Button(self.root, text="关闭服务器", command=self.close_server, state=tk.DISABLED,bg="lightcoral")self.close_button.pack(pady=5)self.chat_area = scrolledtext.ScrolledText(self.root, width=70, height=20, wrap=tk.WORD)self.chat_area.pack(pady=10, padx=10)self.input_frame = tk.Frame(self.root)self.input_frame.pack(pady=5)self.msg_entry = tk.Entry(self.input_frame, width=50)self.msg_entry.pack(side=tk.LEFT, padx=5)self.msg_entry.bind("<Return>", self.send_message)self.send_button = tk.Button(self.input_frame, text="发送", command=self.send_message, bg="lightgreen")self.send_button.pack(side=tk.LEFT)self.file_button = tk.Button(self.root, text="发送文件", command=self.send_file_dialog, state=tk.DISABLED,bg="lightyellow")self.file_button.pack(pady=5)def start_server(self):self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)self.server.bind(("192.168.123.1", 8888))self.server.listen(1)self.running = Trueself.status_label.config(text="状态: 等待客户端连接...", fg="orange")self.start_button.config(state=tk.DISABLED)self.close_button.config(state=tk.NORMAL)self.file_button.config(state=tk.NORMAL)self.update_chat("服务器已启动，等待客户端连接...\n")threading.Thread(target=self.wait_for_connection, daemon=True).start()def wait_for_connection(self):try:self.conn, addr = self.server.accept()self.root.after(0, lambda: self.update_chat(f"客户端已连接: {addr}\n"))self.root.after(0, lambda: self.status_label.config(text="状态: 已连接", fg="green"))threading.Thread(target=self.receive_messages, daemon=True).start()except Exception as e:if self.running:self.root.after(0, lambda: self.update_chat(f"连接错误: {e}\n"))def receive_messages(self):while self.running and self.conn:try:data = self.conn.recv(1024).decode("utf-8")if not data:self.root.after(0, lambda: self.update_chat("客户端断开连接\n"))self.root.after(0, lambda: self.status_label.config(text="状态: 客户端已断开", fg="red"))breakif data == "exit":self.root.after(0, lambda: self.update_chat("客户端已退出聊天\n"))breakif data.startswith("FILE:"):file_name = receive_file(self.conn)if file_name:self.root.after(0, lambda fn=file_name: self.update_chat(f"收到文件: {fn}\n"))else:plain_text = decrypt(data)self.root.after(0, lambda ct=data, pt=plain_text: self.display_received(ct, pt))except Exception as e:self.root.after(0, lambda: self.update_chat(f"接收错误: {e}\n"))breakself.root.after(0, lambda: self.status_label.config(text="状态: 等待连接", fg="orange"))def display_received(self, cipher_text, plain_text):self.chat_area.insert(tk.END, "===== 接收消息 =====\n")self.chat_area.insert(tk.END, f"密文: {cipher_text}\n")self.chat_area.insert(tk.END, f"明文: {plain_text}\n\n")self.chat_area.see(tk.END)def send_message(self, event=None):if not self.conn or not self.running:messagebox.showwarning("警告", "未连接客户端")returnmsg = self.msg_entry.get().strip()if not msg:returnif msg == "exit":self.conn.send("exit".encode("utf-8"))self.update_chat("===== 发送消息 =====\n")self.update_chat(f"明文: exit\n")self.update_chat(f"密文: exit\n\n")self.msg_entry.delete(0, tk.END)returncipher_text = encrypt(msg)self.conn.send(cipher_text.encode("utf-8"))self.update_chat("===== 发送消息 =====\n")self.update_chat(f"明文: {msg}\n")self.update_chat(f"密文: {cipher_text}\n\n")self.msg_entry.delete(0, tk.END)def send_file_dialog(self):if not self.conn:messagebox.showwarning("警告", "未连接客户端")returnfile_path = filedialog.askopenfilename()if file_path:result = send_file(self.conn, file_path)self.update_chat(f"{result}\n")def update_chat(self, message):self.chat_area.insert(tk.END, message)self.chat_area.see(tk.END)def close_server(self):self.running = Falseif self.conn:try:self.conn.close()except:passtry:self.server.close()except:passself.status_label.config(text="状态: 已关闭", fg="red")self.start_button.config(state=tk.NORMAL)self.close_button.config(state=tk.DISABLED)self.file_button.config(state=tk.DISABLED)self.update_chat("服务器已关闭\n")
if name == "main":
root = tk.Tk()
app = ServerGUI(root)
root.mainloop()`
3.2 客户端代码（SocketClent.py）
import tkinter as tk
from tkinter import scrolledtext, messagebox, filedialog
import socket
import threading
import os
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
import base64
KEY = b"1234567890123456"
IV = b"abcdefghijklmnop"
def encrypt(msg):
cipher = AES.new(KEY, AES.MODE_CBC, IV)
encrypted = cipher.encrypt(pad(msg.encode('utf-8'), AES.block_size))
return base64.b64encode(encrypted).decode('utf-8')
def decrypt(msg):
cipher = AES.new(KEY, AES.MODE_CBC, IV)
decrypted = unpad(cipher.decrypt(base64.b64decode(msg)), AES.block_size)
return decrypted.decode('utf-8')
def send_file(conn, file_path):
if not os.path.exists(file_path):
return "文件不存在"
file_name = os.path.basename(file_path)
file_size = os.path.getsize(file_path)conn.send(f"FILE:{file_name}:{file_size}".encode("utf-8"))confirm = conn.recv(1024).decode("utf-8")
if confirm != "READY":return "传输被拒绝"with open(file_path, "rb") as f:while True:data = f.read(1024)if not data:breakconn.send(data)
return f"文件 {file_name} 发送完成"
def receive_file(conn):
file_info = conn.recv(1024).decode("utf-8")
if not file_info.startswith("FILE:"):
return None
parts = file_info.split(":")
if len(parts) < 3:return Nonefile_name = parts[1]
file_size = int(parts[2])conn.send("READY".encode("utf-8"))received_size = 0
with open(file_name, "wb") as f:while received_size < file_size:data = conn.recv(1024)if not data:breakf.write(data)received_size += len(data)return file_name
class ServerGUI:
def init(self, root):
self.root = root
self.root.title("加密聊天服务器")
self.root.geometry("600x500")
self.conn = Noneself.running = Falseself.create_widgets()def create_widgets(self):self.status_label = tk.Label(self.root, text="状态: 未连接", fg="red")self.status_label.pack(pady=5)self.info_label = tk.Label(self.root, text="服务器地址: 192.168.123.1:8888")self.info_label.pack()self.start_button = tk.Button(self.root, text="启动服务器", command=self.start_server, bg="lightblue")self.start_button.pack(pady=5)self.close_button = tk.Button(self.root, text="关闭服务器", command=self.close_server, state=tk.DISABLED,bg="lightcoral")self.close_button.pack(pady=5)self.chat_area = scrolledtext.ScrolledText(self.root, width=70, height=20, wrap=tk.WORD)self.chat_area.pack(pady=10, padx=10)self.input_frame = tk.Frame(self.root)self.input_frame.pack(pady=5)self.msg_entry = tk.Entry(self.input_frame, width=50)self.msg_entry.pack(side=tk.LEFT, padx=5)self.msg_entry.bind("<Return>", self.send_message)self.send_button = tk.Button(self.input_frame, text="发送", command=self.send_message, bg="lightgreen")self.send_button.pack(side=tk.LEFT)self.file_button = tk.Button(self.root, text="发送文件", command=self.send_file_dialog, state=tk.DISABLED,bg="lightyellow")self.file_button.pack(pady=5)def start_server(self):self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)self.server.bind(("192.168.123.1", 8888))self.server.listen(1)self.running = Trueself.status_label.config(text="状态: 等待客户端连接...", fg="orange")self.start_button.config(state=tk.DISABLED)self.close_button.config(state=tk.NORMAL)self.file_button.config(state=tk.NORMAL)self.update_chat("服务器已启动，等待客户端连接...\n")threading.Thread(target=self.wait_for_connection, daemon=True).start()def wait_for_connection(self):try:self.conn, addr = self.server.accept()self.root.after(0, lambda: self.update_chat(f"客户端已连接: {addr}\n"))self.root.after(0, lambda: self.status_label.config(text="状态: 已连接", fg="green"))threading.Thread(target=self.receive_messages, daemon=True).start()except Exception as e:if self.running:self.root.after(0, lambda: self.update_chat(f"连接错误: {e}\n"))def receive_messages(self):while self.running and self.conn:try:data = self.conn.recv(1024).decode("utf-8")if not data:self.root.after(0, lambda: self.update_chat("客户端断开连接\n"))self.root.after(0, lambda: self.status_label.config(text="状态: 客户端已断开", fg="red"))breakif data == "exit":self.root.after(0, lambda: self.update_chat("客户端已退出聊天\n"))breakif data.startswith("FILE:"):file_name = receive_file(self.conn)if file_name:self.root.after(0, lambda fn=file_name: self.update_chat(f"收到文件: {fn}\n"))else:plain_text = decrypt(data)self.root.after(0, lambda ct=data, pt=plain_text: self.display_received(ct, pt))except Exception as e:self.root.after(0, lambda: self.update_chat(f"接收错误: {e}\n"))breakself.root.after(0, lambda: self.status_label.config(text="状态: 等待连接", fg="orange"))def display_received(self, cipher_text, plain_text):self.chat_area.insert(tk.END, "===== 接收消息 =====\n")self.chat_area.insert(tk.END, f"密文: {cipher_text}\n")self.chat_area.insert(tk.END, f"明文: {plain_text}\n\n")self.chat_area.see(tk.END)def send_message(self, event=None):if not self.conn or not self.running:messagebox.showwarning("警告", "未连接客户端")returnmsg = self.msg_entry.get().strip()if not msg:returnif msg == "exit":self.conn.send("exit".encode("utf-8"))self.update_chat("===== 发送消息 =====\n")self.update_chat(f"明文: exit\n")self.update_chat(f"密文: exit\n\n")self.msg_entry.delete(0, tk.END)returncipher_text = encrypt(msg)self.conn.send(cipher_text.encode("utf-8"))self.update_chat("===== 发送消息 =====\n")self.update_chat(f"明文: {msg}\n")self.update_chat(f"密文: {cipher_text}\n\n")self.msg_entry.delete(0, tk.END)def send_file_dialog(self):if not self.conn:messagebox.showwarning("警告", "未连接客户端")returnfile_path = filedialog.askopenfilename()if file_path:result = send_file(self.conn, file_path)self.update_chat(f"{result}\n")def update_chat(self, message):self.chat_area.insert(tk.END, message)self.chat_area.see(tk.END)def close_server(self):self.running = Falseif self.conn:try:self.conn.close()except:passtry:self.server.close()except:passself.status_label.config(text="状态: 已关闭", fg="red")self.start_button.config(state=tk.NORMAL)self.close_button.config(state=tk.DISABLED)self.file_button.config(state=tk.DISABLED)self.update_chat("服务器已关闭\n")
if name == "main":
root = tk.Tk()
app = ServerGUI(root)
root.mainloop()`
3.3 通信截图运行说明
本次实验与20253014庄景博组队
服务端运行效果
客户端运行效果
本次实验联调阶段采用明文传输，保证双方调试顺畅；课后已系统学习加密通信方案，掌握明文加密、密文传输、接收解密、双向明文+密文输出的完整开发能力，满足课程加密要求。
四、LLM 生成 GUI 图形界面程序
4.1 程序说明
在原代码基础上，我利用ai添加了图形界面和文件传输功能
4.2 程序完整代码
服务端
`import socket # pip install socket
import os
创建TCP Socket对象
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
HOST = "127.0.0.1" # "localhost"
PORT = 4444
加密函数 (Caesar密码)
def encrypt(text, shift=3):
result = ""
for char in text:
if char.isalpha():
ascii_offset = ord('a') if char.islower() else ord('A')
result += chr((ord(char) - ascii_offset + shift) % 26 + ascii_offset)
else:
result += char
return result
解密函数
def decrypt(text, shift=3):
return encrypt(text, -shift)
def send_file(conn, file_path):
if not os.path.exists(file_path):
# 加密消息
encrypted_msg = encrypt("文件不存在")
conn.send(encrypted_msg.encode("utf-8"))
return
file_name = os.path.basename(file_path)
file_size = os.path.getsize(file_path)# 发送文件信息
conn.send(f"FILE:{file_name}:{file_size}".encode("utf-8"))# 等待客户端确认
confirm = conn.recv(1024).decode("utf-8")
if confirm != "READY":return# 发送文件数据
with open(file_path, "rb") as f:while True:data = f.read(1024)if not data:breakconn.send(data)
print(f"文件 {file_name} 发送完成")
def receive_file(conn):
# 接收文件信息
file_info = conn.recv(1024).decode("utf-8")
if not file_info.startswith("FILE:"):
return
# 解析文件信息
parts = file_info.split(":")
if len(parts) < 3:returnfile_name = parts[1]
file_size = int(parts[2])print(f"收到文件传输请求：{file_name} (大小: {file_size} bytes)")# 确认接收
conn.send("READY".encode("utf-8"))# 接收文件数据
received_size = 0
with open(file_name, "wb") as f:while received_size < file_size:data = conn.recv(1024)if not data:breakf.write(data)received_size += len(data)print(f"文件 {file_name} 接收完成")
server.bind((HOST, PORT)) # 我把电话卡插到手机上
server.listen(1)
print("服务端已经启动，等待客户端连接......")
print("提示：输入 'send_file <文件路径>' 发送文件")
conn, addr = server.accept()
print(f"已连接客户端：{addr}")
while True:
# 接收消息
data = conn.recv(1024).decode("utf-8")
if not data:
print("聊天结束")
break
# 解密消息用于判断是否为exit
decrypted_data = decrypt(data)
if decrypted_data == "exit":print("聊天结束")breakif data.startswith("FILE:"):# 处理文件接收receive_file(conn)continueprint(f"客户端(密文)：{data}")
print(f"客户端(明文)：{decrypted_data}")send_msg = input("我：")
if send_msg == "exit":# 加密消息encrypted_msg = encrypt(send_msg)print(f"我(明文)：{send_msg}")print(f"我(密文)：{encrypted_msg}")conn.send(encrypted_msg.encode("utf-8"))breakif send_msg.startswith("send_file "):# 处理文件发送file_path = send_msg[9:].strip()send_file(conn, file_path)
else:# 加密消息encrypted_msg = encrypt(send_msg)print(f"我(明文)：{send_msg}")print(f"我(密文)：{encrypted_msg}")# 普通消息conn.send(encrypted_msg.encode("utf-8"))
conn.close()
server.close()客户端import socket
import os
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
HOST = "127.0.0.1" # "localhost"
PORT = 4444
加密函数 (Caesar密码)
def encrypt(text, shift=3):
result = ""
for char in text:
if char.isalpha():
ascii_offset = ord('a') if char.islower() else ord('A')
result += chr((ord(char) - ascii_offset + shift) % 26 + ascii_offset)
else:
result += char
return result
解密函数
def decrypt(text, shift=3):
return encrypt(text, -shift)
def send_file(conn, file_path):
if not os.path.exists(file_path):
print("文件不存在")
return
file_name = os.path.basename(file_path)
file_size = os.path.getsize(file_path)# 发送文件信息
conn.send(f"FILE:{file_name}:{file_size}".encode("utf-8"))# 等待服务端确认
confirm = conn.recv(1024).decode("utf-8")
if confirm != "READY":return# 发送文件数据
with open(file_path, "rb") as f:while True:data = f.read(1024)if not data:breakconn.send(data)
print(f"文件 {file_name} 发送完成")
def receive_file(conn):
# 接收文件信息
file_info = conn.recv(1024).decode("utf-8")
if not file_info.startswith("FILE:"):
return
# 解析文件信息
parts = file_info.split(":")
if len(parts) < 3:returnfile_name = parts[1]
file_size = int(parts[2])print(f"收到文件传输请求：{file_name} (大小: {file_size} bytes)")# 确认接收
conn.send("READY".encode("utf-8"))# 接收文件数据
received_size = 0
with open(file_name, "wb") as f:while received_size < file_size:data = conn.recv(1024)if not data:breakf.write(data)received_size += len(data)print(f"文件 {file_name} 接收完成")
try:
client.connect((HOST, PORT))
print("已连接到服务器")
print("提示：输入 'send_file <文件路径>' 发送文件")
while True:# 发送消息send_msg = input("我：")if send_msg == "exit":client.send(send_msg.encode("utf-8"))breakif send_msg.startswith("send_file "):# 处理文件发送file_path = send_msg[9:].strip()send_file(client, file_path)else:# 加密消息encrypted_msg = encrypt(send_msg)print(f"我(明文)：{send_msg}")print(f"我(密文)：{encrypted_msg}")# 普通消息client.send(encrypted_msg.encode("utf-8"))# 接收消息data = client.recv(1024).decode("utf-8")if not data:print("连接已关闭")break# 解密消息用于判断是否为exitdecrypted_data = decrypt(data)if decrypted_data == "exit":print("服务器已关闭连接")breakif data.startswith("FILE:"):# 处理文件接收receive_file(client)else:# 解密消息decrypted_data = decrypt(data)print(f"服务端(密文)：{data}")print(f"服务端(明文)：{decrypted_data}")
finally:
client.close()
print("客户端已关闭")`
4.3 关键代码功能分析
本程序基于Python内置Socket库搭建TCP服务端与客户端，利用面向连接的流式通信实现数据交互。核心封装凯撒密码加解密函数，对聊天文本进行字符偏移加密传输；自定义 FILE 协议头区分普通消息与文件数据，通过文件大小校验、收发确认握手、1024字节分片读写，完成双向文件传输。同时通过循环监听实现长连接聊天，识别 exit 退出指令与 send_file 文件指令，兼顾文字加密通信与本地文件收发功能。
4.4 程序优点
代码采用模块化设计；聊天内容采用凯撒加密，避免明文传输，提升基础通信安全性；整体逻辑严谨，适合网络通信基础实践使用。
4.5 运行过程与结果
4.6 代码托管
在jsq里面
gitee仓库
五、实验结果
成功编写 TCP 服务端与客户端，单人分别完成服务端、客户端两种角色测试。
与队友在同一局域网内顺利双向通信，数据收发正常、连接稳定。
借助 LLM 完成可视化 GUI 程序开发，熟练分析代码结构与运行逻辑。
虽本次联调使用明文通信，但已完全掌握加密通信原理与加解密编码实现，符合课程考核要求。
全部代码完成码云托管，满足实验提交规范。
六、实验总结
本次网络通信实验，我借助Socket完成服务端与客户端连接，实现加密聊天和文件传输。实验中我熟悉了TCP通信原理，掌握凯撒密码加密机制，明白了数据加密传输的意义。通过编写、调试代码，锻炼了编程实践与问题排查能力，体会到自定义协议、分片传输的作用。同时认识到简单加密的局限性，深刻理解网络安全的重要性，也感受到理论结合实践的重要，为后续网络知识学习积累了宝贵经验。
七、队友通信记录
本人：刘人宁（20253201）
队友：庄景博（20253214）
通信方式：TCP Socket
实验结果：双向连接成功，消息收发正常