Python cryptography库实战：从零构建端到端加密聊天应用

1. 项目概述：为什么我们需要亲手打造一个加密聊天应用？

在数字生活无处不在的今天，聊天应用是我们沟通的血管。但你是否想过，那些看似私密的对话，在传输过程中可能正以“裸奔”的形式穿过无数个网络节点？服务器提供商、网络运营商，甚至是不怀好意的中间人，都有可能窥探到你的信息。这不仅仅是隐私问题，对于商业机密、个人敏感信息的传递，更是一个潜在的风险点。

因此，理解并实践端到端加密（End-to-End Encryption, E2EE）变得至关重要。端到端加密意味着，消息在发送者的设备上就被加密，直到抵达接收者的设备才被解密。在这个过程中，即使是传输消息的服务提供商，也无法读取消息内容。而Python的cryptography库，为我们这些开发者提供了一个绝佳的、相对底层的工具箱，让我们能够亲手实现这一核心安全机制，深刻理解加密通信的每一个环节。

这个项目，就是带你从零开始，用cryptography库构建一个命令行下的加密聊天应用。它不是一个生产级的、拥有华丽UI的应用，而是一个“教学级”的、功能完整的原型。通过它，你将亲手实现：

1. 非对称加密（RSA）：用于安全地交换后续通信所需的对称密钥。
2. 对称加密（AES）：用于高效地加密实际传输的聊天消息。
3. 数字签名：用于验证消息的完整性和发送者身份，防止消息在传输中被篡改或伪造。

这不仅仅是调用几个API，你会深入到密钥生成、序列化、加密模式选择、填充方案等细节。当你完成时，你将获得一个可以真正在本地网络（甚至通过简单的端口转发在互联网上）进行加密聊天的工具，更重要的是，你将透彻理解一个安全通信应用背后的密码学骨架。

2. 核心密码学概念与工具选型解析

在动手写代码之前，我们必须把地基打牢。加密聊天应用的核心是密码学，而cryptography库是我们选择的工具。这里的关键不是死记硬背API，而是理解“为什么”要这么选。

2.1 非对称加密 vs. 对称加密：各司其职的黄金组合

几乎所有现代安全通信协议（如TLS/SSL， Signal协议）都采用混合加密系统，结合了非对称加密和对称加密的优点。

• 非对称加密（如RSA）：有一对密钥，公钥（Public Key）和私钥（Private Key）。公钥可以公开给任何人，用于加密数据；私钥必须严格保密，用于解密用对应公钥加密的数据。它的优点是解决了密钥分发问题，但缺点是计算速度慢，不适合加密大量数据。

  • 在本项目中的应用：我们使用RSA来安全地传递一个“会话密钥”。客户端A生成一个随机的AES对称密钥，然后用服务器B的公钥加密它，再发送给B。只有拥有对应私钥的B才能解密得到这个AES密钥。这样，我们就安全地建立了一个只有A和B知道的共享秘密。

• 对称加密（如AES）：加密和解密使用同一个密钥。它的优点是速度快，适合加密海量数据（如持续的聊天消息流）。

  • 在本项目中的应用：在通过RSA安全交换了AES密钥之后，后续所有的聊天消息都使用这个AES密钥进行加密和解密。这保证了通信的高效性。

注意：直接使用RSA加密长消息是错误且不安全的做法。RSA有长度限制，并且性能低下。正确的模式永远是：用RSA加密一个随机的对称密钥，然后用这个对称密钥去加密实际数据。

2.2 加密模式与填充：让AES更安全

选择了AES，事情还没完。AES是一个分组密码算法，它一次只能处理固定长度（128位，即16字节）的一块数据。我们的消息长度是随机的，这就需要“模式”和“填充”。

• 加密模式：我们选择GCM（Galois/Counter Mode）。这是现代应用中的首选，原因有三：

  1. 认证加密：GCM模式不仅提供保密性（加密），还同时提供完整性认证。它会生成一个“认证标签”（Tag），接收方可以验证密文在传输中是否被篡改。这比传统的CBC模式需要单独计算HMAC要方便和高效。
  2. 无需填充：GCM是一种流密码模式，它不需要将数据填充到固定的块长度，避免了填充预言攻击等风险。
  3. 支持附加数据：可以关联加密一些不需要保密但需要认证的附加数据（如消息头、发送者ID），非常灵活。

• 为什么不用ECB或CBC？ECB模式极其不安全，相同的明文块会产生相同的密文块，会泄露模式信息。CBC模式需要填充，且容易受到填充预言攻击，除非实现得非常小心。因此，对于新项目，GCM是更安全、更推荐的选择。

2.3 数字签名：证明“你是谁”和“消息没被改”

加密保证了消息内容不被窥探，但还需要防止攻击者冒充他人发送消息，或者篡改密文（虽然GCM能检测篡改，但签名提供了不可否认性）。

我们使用RSA 密钥对 + PSS 填充方案来进行签名和验证。

• 发送方：用自己的私钥对消息的哈希值（如SHA-256）进行签名，将签名和消息（或消息密文）一起发送。
• 接收方：用发送方的公钥验证签名。如果验证通过，则证明：1) 消息确实来自声称的发送者（因为只有他有私钥）；2) 消息在签名后未被更改。

2.4 Python cryptography库：为什么是它？

Python有很多加密库，如pycryptodome、M2Crypto等。cryptography库有其独特优势：

• 底层基于稳健的C库：它通常是OpenSSL或LibreSSL的绑定，这意味着其核心加密操作经过了长时间、高强度的安全审计和实践检验，比纯Python实现的算法更安全、更快。
• “Hazmat”分层设计：库分为两层：“安全配方”层（cryptography.fernet,cryptography.hazmat.primitives.asymmetric中的高级接口）和“危险材料”层（cryptography.hazmat.primitives）。高级接口更安全、易用，但不够灵活。我们的项目需要深入细节，因此会接触“危险材料”层，这要求我们更清楚自己在做什么。
• API设计清晰：虽然深入底层，但其API设计相对清晰，文档也较为完善，适合学习。

3. 项目架构设计与核心模块拆解

我们将应用分为客户端和服务器端，两者代码结构相似，但角色不同。为了简化，我们让服务器也充当一个聊天用户。在实际运行中，你需要启动一个服务器进程和至少一个客户端进程。

加密聊天应用架构 ├── 核心密码学模块 (crypto_utils.py) │ ├── 生成RSA密钥对 │ ├── 序列化/反序列化密钥 (PEM格式) │ ├── 使用RSA公钥加密数据 │ ├── 使用RSA私钥解密数据 │ ├── 使用RSA私钥签名数据 │ ├── 使用RSA公钥验证签名 │ ├── 生成随机AES密钥 (用于GCM) │ ├── 使用AES-GCM加密消息 │ └── 使用AES-GCM解密消息 ├── 服务器端 (server.py) │ ├── 加载或生成服务器自身的RSA密钥对 │ ├── 绑定Socket，监听客户端连接 │ ├── 接受连接，接收客户端公钥 │ ├── 发送服务器公钥给客户端 │ ├── 接收客户端发来的加密会话密钥，并用服务器私钥解密 │ ├── 进入加密聊天循环： │ │ ├── 接收加密消息和签名 │ │ ├── 验证签名 │ │ ├── 解密消息并显示 │ │ ├── 读取用户输入，加密并签名后发送 │ └── 处理连接断开 └── 客户端端 (client.py) ├── 加载或生成客户端自身的RSA密钥对 ├── 连接服务器Socket ├── 交换公钥 ├── 生成随机AES会话密钥，用服务器公钥加密后发送 ├── 进入加密聊天循环（同服务器端） └── 处理连接断开

通信协议设计（简化版）： 我们设计一个简单的二进制协议来区分不同类型的消息。每条网络消息都有一个固定的消息头。

# 伪代码表示消息结构 class Message: type: int (1字节， 0=公钥， 1=加密的会话密钥， 2=聊天消息) length: int (4字节， 网络字节序， 表示数据部分长度) data: bytes (实际数据， 长度由`length`指定)

例如，类型为2的聊天消息，其data部分可能是一个JSON字符串，包含了加密的密文、GCM的Nonce和Tag，以及发送者的数字签名。

4. 核心密码学工具函数实现（crypto_utils.py）

这是整个项目的心脏。我们将所有密码学操作封装在这里，确保主程序逻辑清晰。

4.1 生成与处理RSA密钥

from cryptography.hazmat.primitives.asymmetric import rsa, padding from cryptography.hazmat.primitives import serialization, hashes from cryptography.hazmat.backends import default_backend import os def generate_rsa_keypair(key_size=2048): """ 生成RSA公私钥对。 2048位是当前安全的最小推荐值。4096位更安全，但生成和使用稍慢。 """ private_key = rsa.generate_private_key( public_exponent=65537, # 标准公钥指数 key_size=key_size, backend=default_backend() ) public_key = private_key.public_key() return private_key, public_key def serialize_public_key(public_key): """将公钥序列化为PEM格式的字节串，便于网络传输或存储。""" return public_key.public_bytes( encoding=serialization.Encoding.PEM, format=serialization.PublicFormat.SubjectPublicKeyInfo ) def serialize_private_key(private_key, password=None): """将私钥序列化为PEM格式。可以可选地用密码加密。""" encryption_algorithm = serialization.NoEncryption() if password: encryption_algorithm = serialization.BestAvailableEncryption(password.encode()) return private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=encryption_algorithm ) def deserialize_public_key(public_key_bytes): """从PEM格式的字节串加载公钥。""" return serialization.load_pem_public_key(public_key_bytes, backend=default_backend()) def deserialize_private_key(private_key_bytes, password=None): """从PEM格式的字节串加载私钥。如果私钥被加密，需要提供密码。""" password_bytes = password.encode() if password else None return serialization.load_pem_private_key( private_key_bytes, password=password_bytes, backend=default_backend() )

4.2 RSA加密、解密、签名与验证

def rsa_encrypt(public_key, plaintext): """使用RSA公钥加密数据。用于加密短的会话密钥。""" # 使用OAEP填充，这是目前推荐的、抵抗各种攻击的填充方案。 ciphertext = public_key.encrypt( plaintext, padding.OAEP( mgf=padding.MGF1(algorithm=hashes.SHA256()), algorithm=hashes.SHA256(), label=None # 通常为None ) ) return ciphertext def rsa_decrypt(private_key, ciphertext): """使用RSA私钥解密数据。""" plaintext = private_key.decrypt( ciphertext, padding.OAEP( mgf=padding.MGF1(algorithm=hashes.SHA256()), algorithm=hashes.SHA256(), label=None ) ) return plaintext def sign_message(private_key, message): """使用RSA私钥对消息进行签名。""" # 先对消息做哈希，然后对哈希值签名。使用PSS填充方案。 signature = private_key.sign( message, padding.PSS( mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.MAX_LENGTH # 使用最大盐值长度增强安全性 ), hashes.SHA256() ) return signature def verify_signature(public_key, message, signature): """使用RSA公钥验证消息签名。""" try: public_key.verify( signature, message, padding.PSS( mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.MAX_LENGTH ), hashes.SHA256() ) return True # 验证成功 except Exception as e: # 通常是InvalidSignature异常 # 在实际应用中，最好只捕获InvalidSignature，这里简化处理 return False # 验证失败

4.3 AES-GCM对称加密与解密

from cryptography.hazmat.primitives.ciphers.aead import AESGCM import secrets # 用于生成密码学安全的随机数 def generate_aes_key(): """生成一个随机的256位（32字节）AES密钥，用于GCM模式。""" return secrets.token_bytes(32) # AES-256 def aes_gcm_encrypt(key, plaintext, associated_data=b""): """ 使用AES-GCM加密数据。 :param key: 32字节的AES密钥 :param plaintext: 需要加密的明文 :param associated_data: 需要认证但不加密的附加数据 :return: (nonce, ciphertext, tag) 三元组 """ # 生成一个随机的96位（12字节）Nonce。每次加密都必须使用新的Nonce！ nonce = secrets.token_bytes(12) aesgcm = AESGCM(key) # 加密并生成认证标签。结果是一个字节串，包含密文和标签。 encrypted_data = aesgcm.encrypt(nonce, plaintext, associated_data) # 通常，密文和标签是连在一起的。AESGCM加密返回 ciphertext + tag。 # 对于256位密钥和96位nonce，tag长度是16字节。 ciphertext_with_tag = encrypted_data # 为了方便，我们返回nonce和完整的加密结果。 # 另一种常见做法是分离tag，这里我们保持库返回的原样。 return nonce, ciphertext_with_tag def aes_gcm_decrypt(key, nonce, ciphertext_with_tag, associated_data=b""): """ 使用AES-GCM解密数据。 :param key: 32字节的AES密钥 :param nonce: 加密时使用的12字节Nonce :param ciphertext_with_tag: 密文和认证标签的组合体 :param associated_data: 加密时使用的附加数据 :return: 解密后的明文，如果认证失败会抛出InvalidTag异常 """ aesgcm = AESGCM(key) try: plaintext = aesgcm.decrypt(nonce, ciphertext_with_tag, associated_data) return plaintext except Exception as e: # 这里最可能的是InvalidTag异常，表示密文被篡改或密钥错误。 print(f"解密失败或认证失败: {e}") return None

实操心得：secrets.token_bytes()比os.urandom()在语义上更清晰，它是Python 3.6+引入的专门用于生成密码学安全随机数的模块。对于Nonce和密钥的生成，务必使用它。绝对不要使用random模块，它是伪随机的，不安全。

5. 网络通信与协议实现

有了密码学工具，我们需要一个可靠的网络通道来传输这些加密后的数据。我们使用Python内置的socket和struct模块来实现简单的二进制协议。

5.1 消息编解码器

# network_utils.py import struct import json from enum import IntEnum class MessageType(IntEnum): PUBLIC_KEY = 0 ENCRYPTED_SESSION_KEY = 1 CHAT_MESSAGE = 2 # 可以扩展其他类型，如文件传输、命令等 def encode_message(msg_type, data): """ 将消息类型和数据编码为网络传输的字节流。 格式: [1字节类型][4字节数据长度][N字节数据] """ if not isinstance(data, bytes): # 如果不是字节，尝试序列化为JSON再编码 data = json.dumps(data).encode('utf-8') length = len(data) # ! 表示网络字节序（大端序），B表示1字节无符号整数，I表示4字节无符号整数 header = struct.pack('!BI', msg_type, length) return header + data def decode_message_header(sock): """ 从socket中读取消息头（5字节）。 返回 (msg_type, length)。 如果连接关闭，返回 (None, None)。 """ header = sock.recv(5) # 先读取固定的5字节头 if len(header) < 5: # 连接可能已关闭 return None, None msg_type, length = struct.unpack('!BI', header) return msg_type, length def receive_all(sock, length): """ 从socket中精确接收指定长度的数据。 这是处理TCP流式传输的关键，因为recv可能一次收不满。 """ data = b'' while len(data) < length: packet = sock.recv(length - len(data)) if not packet: # 连接中断 return None data += packet return data

5.2 服务器端主循环实现

服务器端的主要职责是监听连接、进行密钥交换、然后中继加密的聊天消息。在我们的点对点模型中，服务器也作为一个聊天参与者。

# server.py (核心部分) import socket import threading from crypto_utils import * from network_utils import * class ChatServer: def __init__(self, host='0.0.0.0', port=5555): self.host = host self.port = port self.server_private_key, self.server_public_key = generate_rsa_keypair() # 存储客户端信息： socket, 公钥, 会话密钥 self.clients = {} def start(self): server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server_socket.bind((self.host, self.port)) server_socket.listen(5) print(f"[*] 服务器启动，监听 {self.host}:{self.port}") while True: client_socket, addr = server_socket.accept() print(f"[+] 新连接来自 {addr}") # 为每个客户端创建一个新线程处理 client_handler = threading.Thread( target=self.handle_client, args=(client_socket, addr) ) client_handler.start() def handle_client(self, client_socket, addr): """处理单个客户端的完整会话。""" try: # 1. 发送服务器公钥给客户端 pub_key_bytes = serialize_public_key(self.server_public_key) client_socket.send(encode_message(MessageType.PUBLIC_KEY, pub_key_bytes)) # 2. 接收客户端公钥 msg_type, length = decode_message_header(client_socket) if msg_type != MessageType.PUBLIC_KEY: print(f"[-] {addr} 协议错误，期望公钥") return client_pub_key_bytes = receive_all(client_socket, length) client_public_key = deserialize_public_key(client_pub_key_bytes) # 3. 接收客户端发来的加密会话密钥 msg_type, length = decode_message_header(client_socket) if msg_type != MessageType.ENCRYPTED_SESSION_KEY: print(f"[-] {addr} 协议错误，期望加密会话密钥") return encrypted_session_key = receive_all(client_socket, length) # 用服务器私钥解密，得到AES会话密钥 session_key = rsa_decrypt(self.server_private_key, encrypted_session_key) # 存储客户端信息 self.clients[addr] = { 'socket': client_socket, 'public_key': client_public_key, 'session_key': session_key } print(f"[*] {addr} 密钥交换完成，会话已加密。") # 4. 进入加密聊天循环 self.chat_loop(client_socket, addr, session_key, client_public_key) except Exception as e: print(f"[-] 处理客户端 {addr} 时发生错误: {e}") finally: client_socket.close() if addr in self.clients: del self.clients[addr] print(f"[-] {addr} 连接已关闭。") def chat_loop(self, sock, addr, session_key, peer_public_key): """加密聊天主循环。""" while True: # 接收消息 msg_type, length = decode_message_header(sock) if msg_type is None: # 连接关闭 break if msg_type != MessageType.CHAT_MESSAGE: print(f"[!] {addr} 收到非聊天消息类型，忽略。") # 跳过该消息数据 receive_all(sock, length) continue encrypted_packet = receive_all(sock, length) if encrypted_packet is None: break # 解析数据包 (假设我们传输的是JSON，包含nonce, ciphertext_with_tag, signature) try: packet_data = json.loads(encrypted_packet.decode('utf-8')) nonce = bytes.fromhex(packet_data['nonce']) ciphertext_with_tag = bytes.fromhex(packet_data['ciphertext']) signature = bytes.fromhex(packet_data['signature']) if packet_data.get('signature') else None except (json.JSONDecodeError, KeyError) as e: print(f"[!] {addr} 消息格式错误: {e}") continue # 解密消息 plaintext = aes_gcm_decrypt(session_key, nonce, ciphertext_with_tag) if plaintext is None: print(f"[!] {addr} 消息解密或认证失败！") continue # 验证签名 (如果提供了签名) if signature and peer_public_key: if not verify_signature(peer_public_key, ciphertext_with_tag, signature): print(f"[!] {addr} 消息签名验证失败！消息可能被篡改或伪造。") # 可以选择断开连接或仅警告 # continue else: print(f"[+] {addr} 消息签名验证通过。") # 显示消息 try: message = plaintext.decode('utf-8') print(f"\n[来自 {addr}]> {message}") except UnicodeDecodeError: print(f"\n[来自 {addr}]> (收到非文本消息或解码错误)") # 发送回复 (这里简化处理，服务器端手动输入) # 在实际点对点聊天中，这里应该是另一个线程接收本地输入并发送。 # 为了演示，我们简化：服务器收到消息后，可以手动输入回复。 # 更完整的实现需要多线程处理I/O。 # 以下代码块在单线程下会阻塞接收，仅作示意。 # reply = input(f"[发送给 {addr}]> ") # if reply.lower() == 'exit': # break # self._send_encrypted_message(sock, session_key, self.server_private_key, reply)

5.3 客户端端主循环实现

客户端逻辑与服务器端镜像，负责发起连接、发起密钥交换、发送和接收消息。

# client.py (核心部分) import socket import threading import sys from crypto_utils import * from network_utils import * class ChatClient: def __init__(self, server_host='127.0.0.1', server_port=5555): self.server_host = server_host self.server_port = server_port self.client_private_key, self.client_public_key = generate_rsa_keypair() self.session_key = None # 与服务器共享的AES密钥 self.server_public_key = None def connect(self): """连接到服务器并完成密钥交换。""" self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: self.sock.connect((self.server_host, self.server_port)) print(f"[*] 已连接到服务器 {self.server_host}:{self.server_port}") except ConnectionRefusedError: print(f"[-] 无法连接到服务器，请检查地址和端口，并确保服务器已启动。") sys.exit(1) # 1. 接收服务器公钥 msg_type, length = decode_message_header(self.sock) if msg_type != MessageType.PUBLIC_KEY: print(f"[-] 协议错误：期望收到服务器公钥") self.sock.close() return False server_pub_key_bytes = receive_all(self.sock, length) self.server_public_key = deserialize_public_key(server_pub_key_bytes) # 2. 发送客户端公钥给服务器 client_pub_key_bytes = serialize_public_key(self.client_public_key) self.sock.send(encode_message(MessageType.PUBLIC_KEY, client_pub_key_bytes)) # 3. 生成并发送加密的会话密钥 self.session_key = generate_aes_key() encrypted_session_key = rsa_encrypt(self.server_public_key, self.session_key) self.sock.send(encode_message(MessageType.ENCRYPTED_SESSION_KEY, encrypted_session_key)) print("[+] 密钥交换完成，加密通道已建立。") return True def start_chat(self): """启动聊天，创建两个线程分别处理接收和发送。""" if not self.session_key: print("[-] 尚未建立加密会话。") return # 接收消息线程 recv_thread = threading.Thread(target=self._receive_messages, daemon=True) recv_thread.start() # 发送消息线程 (主线程) print("\n你可以开始输入消息了。输入 'exit' 退出。\n") self._send_messages() def _receive_messages(self): """持续接收并解密来自服务器的消息。""" while True: try: msg_type, length = decode_message_header(self.sock) if msg_type is None: print("\n[-] 与服务器的连接已断开。") break if msg_type != MessageType.CHAT_MESSAGE: # 忽略其他类型消息，或做相应处理 data = receive_all(self.sock, length) continue encrypted_packet = receive_all(self.sock, length) if encrypted_packet is None: break packet_data = json.loads(encrypted_packet.decode('utf-8')) nonce = bytes.fromhex(packet_data['nonce']) ciphertext_with_tag = bytes.fromhex(packet_data['ciphertext']) signature = bytes.fromhex(packet_data['signature']) if packet_data.get('signature') else None plaintext = aes_gcm_decrypt(self.session_key, nonce, ciphertext_with_tag) if plaintext is None: print("[!] 收到一条无法解密或认证失败的消息。") continue if signature and self.server_public_key: if not verify_signature(self.server_public_key, ciphertext_with_tag, signature): print("[!] 警告：收到一条签名验证失败的消息！") # else: 验证成功，安静处理 message = plaintext.decode('utf-8') print(f"\n[服务器] > {message}") # 打印接收到的消息 # 打印提示符，避免和输入行混淆。这是一个简单的处理。 sys.stdout.write("[你] > ") sys.stdout.flush() except (ConnectionResetError, json.JSONDecodeError, KeyError) as e: print(f"\n[-] 接收消息时出错: {e}") break def _send_messages(self): """读取用户输入，加密并发送消息。""" while True: try: message = input("[你] > ") # 注意：在多线程下，input可能会被接收线程的输出打断。 if message.lower() == 'exit': print("[*] 退出聊天。") self.sock.close() break if not message.strip(): continue # 加密并签名消息 nonce, ciphertext_with_tag = aes_gcm_encrypt(self.session_key, message.encode('utf-8')) signature = sign_message(self.client_private_key, ciphertext_with_tag) # 构建传输包 packet = { 'nonce': nonce.hex(), 'ciphertext': ciphertext_with_tag.hex(), 'signature': signature.hex() } packet_bytes = json.dumps(packet).encode('utf-8') self.sock.send(encode_message(MessageType.CHAT_MESSAGE, packet_bytes)) except (BrokenPipeError, ConnectionResetError): print("[-] 连接已断开，无法发送消息。") break except Exception as e: print(f"[-] 发送消息时出错: {e}") break if __name__ == "__main__": client = ChatClient() if client.connect(): client.start_chat()

6. 运行、测试与常见问题排查

6.1 如何运行这个应用？

1. 准备环境：确保你安装了Python 3.6+，然后安装cryptography库。

   pip install cryptography

2. 保存代码：将上述代码块分别保存为crypto_utils.py、network_utils.py、server.py、client.py。
3. 启动服务器：在一个终端窗口运行。

   python server.py

   你会看到类似[*] 服务器启动，监听 0.0.0.0:5555的输出。
4. 启动客户端：在另一个终端窗口运行。

   python client.py

   默认连接本地服务器(127.0.0.1:5555)。如果需要连接其他机器，可以修改client.py中的server_host参数。
5. 开始聊天：客户端连接成功后，双方即可在各自的终端输入文字进行加密聊天。输入exit退出客户端。

6.2 典型问题与解决方案速查表

6.3 安全注意事项与进阶思考

1. 本项目是教学原型：它缺少许多生产级应用必需的安全特性，例如：

   • 前向保密：每次会话使用相同的RSA密钥对。如果服务器私钥长期泄露，所有过去的会话记录都可能被解密。生产环境应使用像Diffie-Hellman这样的密钥交换协议来实现前向保密。
   • 身份认证：我们只是交换了公钥，但没有验证对方公钥的真实性（即“这真的是Alice的公钥吗？”）。这容易受到中间人攻击。需要引入证书、信任链或“密钥指纹”验证（如Signal的Safety Number）。
   • 消息重放攻击防护：虽然GCM的Nonce重复使用是灾难性的，但我们没有机制防止攻击者记录并重复发送一条有效的加密消息。需要引入消息序号或时间戳。
   • 完善的密钥管理：密钥应安全存储（如使用操作系统密钥链），而不是每次运行都重新生成。

2. Nonce的管理至关重要：AES-GCM中，同一个密钥下，绝对不允许重复使用Nonce。一旦重复，安全性将彻底崩溃。我们每次加密都生成随机Nonce是正确做法。在生产系统中，有时会使用计数器作为Nonce的一部分以确保唯一性。

3. 使用现成的协议：对于真实项目，强烈建议不要自己从头设计加密协议。应使用经过严格审计的现有库和协议，如：

   • TLS：用于保护客户端-服务器通信。你的聊天应用可以直接建立在TLS连接之上，省去大量底层密码学实现。
   • Signal协议：端到端加密的黄金标准，由Open Whisper Systems设计，被WhatsApp、Signal等广泛应用。有优秀的开源实现（如libsignal）。
   • 成熟的加密库：如libsodium（PyNaCl是其Python绑定），它提供了更高级、更易用且更安全的API（如crypto_box用于非对称加密，crypto_secretbox用于对称加密）。

亲手实现这个项目最大的价值，在于剥开了现代加密通信应用的神秘外衣，让你理解了“混合加密”、“认证加密”、“数字签名”这些概念是如何在代码层面协同工作的。下次当你使用一个标榜“端到端加密”的应用时，你就能大致想象出它背后正在运行的、类似但远比你今天构建的更加复杂和严谨的密码学机器。