基于Python開(kāi)發(fā)簡(jiǎn)易局域網(wǎng)聊天工具
更新時(shí)間：2025年07月28日 09:02:27 作者：超級(jí)小識(shí)
端對(duì)端加密技術(shù)可以確保只有通信的雙方能夠讀取消息內(nèi)容,即使是服務(wù)器也無(wú)法解密,本文將詳細(xì)介紹如何用Python實(shí)現(xiàn)一個(gè)簡(jiǎn)單的局域網(wǎng)聊天工具,并為其添加端對(duì)端加密功能,感興趣的小伙伴可以了解下
在現(xiàn)代互聯(lián)網(wǎng)環(huán)境中，隱私和安全越來(lái)越受到重視。端對(duì)端加密（End-to-End Encryption, E2EE）技術(shù)可以確保只有通信的雙方能夠讀取消息內(nèi)容，即使是服務(wù)器也無(wú)法解密。本文將詳細(xì)介紹如何用Python實(shí)現(xiàn)一個(gè)簡(jiǎn)單的局域網(wǎng)聊天工具，并為其添加端對(duì)端加密功能。
什么是端對(duì)端加密
端對(duì)端加密是一種通信加密方式，消息在發(fā)送端加密后，只有接收端能夠解密。中間的任何節(jié)點(diǎn)（如路由器、服務(wù)器等）都只能看到加密后的數(shù)據(jù)，無(wú)法獲取原始消息內(nèi)容。這種加密方式廣泛應(yīng)用于即時(shí)通訊工具中，如WhatsApp、Signal等。
項(xiàng)目概述
本項(xiàng)目將實(shí)現(xiàn)以下功能：
- 基于Python的局域網(wǎng)聊天工具
- 支持多客戶端連接
- 使用非對(duì)稱加密（RSA）進(jìn)行密鑰交換
- 使用對(duì)稱加密（AES）加密通信內(nèi)容
- 簡(jiǎn)單的命令行界面
技術(shù)棧
編程語(yǔ)言：Python 3.x
網(wǎng)絡(luò)庫(kù)：socket、threading
加密庫(kù)：cryptography
其他：argparse（參數(shù)解析）
核心模塊解析
網(wǎng)絡(luò)通信基礎(chǔ)
局域網(wǎng)聊天工具的核心是網(wǎng)絡(luò)通信。Python的socket庫(kù)提供了低級(jí)別的網(wǎng)絡(luò)接口，可以創(chuàng)建TCP或UDP連接。本項(xiàng)目中采用TCP協(xié)議，因?yàn)樗鼙ＷC消息的可靠傳輸。
import socket # 創(chuàng)建TCP socket server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
TCP通信需要明確的客戶端和服務(wù)器端。服務(wù)器監(jiān)聽(tīng)特定端口，客戶端主動(dòng)連接服務(wù)器。在聊天工具中，每個(gè)用戶既是客戶端（發(fā)送消息）又是服務(wù)器（接收消息）。
多線程處理
為了實(shí)現(xiàn)同時(shí)接收和發(fā)送消息，需要使用多線程。Python的threading模塊可以方便地創(chuàng)建和管理線程。
import threading
def receive_messages():
while True:
data = client_socket.recv(1024)
print(f"Received: {data.decode()}")
# 創(chuàng)建接收消息的線程
receive_thread = threading.Thread(target=receive_messages)
receive_thread.start()
加密實(shí)現(xiàn)
加密部分分為兩個(gè)階段：密鑰交換和消息加密。
密鑰交換：使用RSA非對(duì)稱加密。每個(gè)用戶生成自己的RSA密鑰對(duì)，公開(kāi)公鑰，保存私鑰。當(dāng)兩個(gè)用戶通信時(shí)，他們交換公鑰，然后用對(duì)方的公鑰加密一個(gè)隨機(jī)的AES密鑰（會(huì)話密鑰）。
消息加密：使用AES對(duì)稱加密。一旦會(huì)話密鑰安全交換，后續(xù)通信都使用這個(gè)密鑰進(jìn)行加密解密，因?yàn)閷?duì)稱加密比非對(duì)稱加密效率高很多。
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives import serialization, hashes
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
import os
# 生成RSA密鑰對(duì)
private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend()
)
public_key = private_key.public_key()
# 生成AES密鑰
aes_key = os.urandom(32) # 256-bit key
詳細(xì)實(shí)現(xiàn)步驟
用戶類設(shè)計(jì)
首先設(shè)計(jì)一個(gè)User類，包含用戶的基本信息和加密相關(guān)操作。
class User:
def __init__(self, name):
self.name = name
self.private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend()
)
self.public_key = self.private_key.public_key()
self.peer_public_key = None
self.aes_key = None
def serialize_public_key(self):
return self.public_key.public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo
)
def deserialize_public_key(self, key_bytes):
self.peer_public_key = serialization.load_pem_public_key(
key_bytes,
backend=default_backend()
)
密鑰交換協(xié)議
設(shè)計(jì)一個(gè)簡(jiǎn)單的協(xié)議來(lái)交換公鑰和會(huì)話密鑰：
- 連接建立后，雙方交換RSA公鑰
- 一方生成AES密鑰，用對(duì)方的公鑰加密后發(fā)送
- 對(duì)方收到后用私鑰解密獲取AES密鑰
- 后續(xù)通信使用AES加密
def perform_key_exchange(user, conn, is_initiator):
# 交換公鑰
conn.sendall(user.serialize_public_key())
peer_key_bytes = conn.recv(4096)
user.deserialize_public_key(peer_key_bytes)
if is_initiator:
# 生成并發(fā)送AES密鑰
user.aes_key = os.urandom(32)
encrypted_aes_key = user.peer_public_key.encrypt(
user.aes_key,
padding.OAEP(
mgf=padding.MGF1(algorithm=hashes.SHA256()),
algorithm=hashes.SHA256(),
label=None
)
)
conn.sendall(encrypted_aes_key)
else:
# 接收并解密AES密鑰
encrypted_aes_key = conn.recv(4096)
user.aes_key = user.private_key.decrypt(
encrypted_aes_key,
padding.OAEP(
mgf=padding.MGF1(algorithm=hashes.SHA256()),
algorithm=hashes.SHA256(),
label=None
)
)
消息加密解密
實(shí)現(xiàn)AES加密解密功能：
def encrypt_message(key, message):
iv = os.urandom(16) # 初始向量
cipher = Cipher(
algorithms.AES(key),
modes.CFB(iv),
backend=default_backend()
)
encryptor = cipher.encryptor()
ciphertext = encryptor.update(message.encode()) + encryptor.finalize()
return iv + ciphertext
def decrypt_message(key, ciphertext):
iv = ciphertext[:16]
cipher = Cipher(
algorithms.AES(key),
modes.CFB(iv),
backend=default_backend()
)
decryptor = cipher.decryptor()
plaintext = decryptor.update(ciphertext[16:]) + decryptor.finalize()
return plaintext.decode()
服務(wù)器和客戶端實(shí)現(xiàn)
將上述功能整合到服務(wù)器和客戶端代碼中：
def start_server(port):
user = User("Server")
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(('0.0.0.0', port))
s.listen()
conn, addr = s.accept()
perform_key_exchange(user, conn, False)
# 啟動(dòng)接收消息線程
def receive():
while True:
data = conn.recv(4096)
if not data: break
message = decrypt_message(user.aes_key, data)
print(f"Received: {message}")
threading.Thread(target=receive, daemon=True).start()
# 發(fā)送消息
while True:
message = input("> ")
encrypted = encrypt_message(user.aes_key, message)
conn.sendall(encrypted)
def start_client(host, port):
user = User("Client")
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.connect((host, port))
perform_key_exchange(user, s, True)
# 啟動(dòng)接收消息線程
def receive():
while True:
data = s.recv(4096)
if not data: break
message = decrypt_message(user.aes_key, data)
print(f"Received: {message}")
threading.Thread(target=receive, daemon=True).start()
# 發(fā)送消息
while True:
message = input("> ")
encrypted = encrypt_message(user.aes_key, message)
s.sendall(encrypted)
完整源代碼
import socket
import threading
import argparse
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives import serialization, hashes
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
import os
class User:
def __init__(self, name):
self.name = name
self.private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend()
)
self.public_key = self.private_key.public_key()
self.peer_public_key = None
self.aes_key = None
def serialize_public_key(self):
return self.public_key.public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo
)
def deserialize_public_key(self, key_bytes):
self.peer_public_key = serialization.load_pem_public_key(
key_bytes,
backend=default_backend()
)
def encrypt_message(key, message):
iv = os.urandom(16)
cipher = Cipher(
algorithms.AES(key),
modes.CFB(iv),
backend=default_backend()
)
encryptor = cipher.encryptor()
ciphertext = encryptor.update(message.encode()) + encryptor.finalize()
return iv + ciphertext
def decrypt_message(key, ciphertext):
iv = ciphertext[:16]
cipher = Cipher(
algorithms.AES(key),
modes.CFB(iv),
backend=default_backend()
)
decryptor = cipher.decryptor()
plaintext = decryptor.update(ciphertext[16:]) + decryptor.finalize()
return plaintext.decode()
def perform_key_exchange(user, conn, is_initiator):
conn.sendall(user.serialize_public_key())
peer_key_bytes = conn.recv(4096)
user.deserialize_public_key(peer_key_bytes)
if is_initiator:
user.aes_key = os.urandom(32)
encrypted_aes_key = user.peer_public_key.encrypt(
user.aes_key,
padding.OAEP(
mgf=padding.MGF1(algorithm=hashes.SHA256()),
algorithm=hashes.SHA256(),
label=None
)
)
conn.sendall(encrypted_aes_key)
else:
encrypted_aes_key = conn.recv(4096)
user.aes_key = user.private_key.decrypt(
encrypted_aes_key,
padding.OAEP(
mgf=padding.MGF1(algorithm=hashes.SHA256()),
algorithm=hashes.SHA256(),
label=None
)
)
def start_server(port):
user = User("Server")
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(('0.0.0.0', port))
s.listen()
print(f"Server listening on port {port}")
conn, addr = s.accept()
print(f"Connected by {addr}")
perform_key_exchange(user, conn, False)
def receive():
while True:
data = conn.recv(4096)
if not data: break
message = decrypt_message(user.aes_key, data)
print(f"Received: {message}")
threading.Thread(target=receive, daemon=True).start()
while True:
message = input("> ")
encrypted = encrypt_message(user.aes_key, message)
conn.sendall(encrypted)
def start_client(host, port):
user = User("Client")
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.connect((host, port))
perform_key_exchange(user, s, True)
def receive():
while True:
data = s.recv(4096)
if not data: break
message = decrypt_message(user.aes_key, data)
print(f"Received: {message}")
threading.Thread(target=receive, daemon=True).start()
while True:
message = input("> ")
encrypted = encrypt_message(user.aes_key, message)
s.sendall(encrypted)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Secure LAN Chat")
parser.add_argument("-s", "--server", action="store_true", help="Run as server")
parser.add_argument("-c", "--client", action="store_true", help="Run as client")
parser.add_argument("--host", type=str, default="localhost", help="Server host")
parser.add_argument("--port", type=int, default=12345, help="Port number")
args = parser.parse_args()
if args.server:
start_server(args.port)
elif args.client:
start_client(args.host, args.port)
else:
print("Please specify --server or --client")
使用說(shuō)明
在一臺(tái)機(jī)器上運(yùn)行服務(wù)器：
python chat.py --server --port 12345
在另一臺(tái)機(jī)器上運(yùn)行客戶端（確保在同一局域網(wǎng)）：
python chat.py --client --host <服務(wù)器IP> --port 12345
開(kāi)始聊天，輸入的消息會(huì)自動(dòng)加密傳輸
安全注意事項(xiàng)
- 本項(xiàng)目?jī)H用于學(xué)習(xí)目的，不應(yīng)用于生產(chǎn)環(huán)境
- 實(shí)際應(yīng)用中需要更完善的密鑰管理和身份驗(yàn)證機(jī)制
- 加密實(shí)現(xiàn)使用了cryptography庫(kù)，這是Python中比較可靠的加密庫(kù)
- 確保Python環(huán)境是最新版本，避免已知的安全漏洞
通過(guò)這個(gè)項(xiàng)目，您可以學(xué)習(xí)到網(wǎng)絡(luò)編程、多線程、加密算法等多項(xiàng)技術(shù)。
到此這篇關(guān)于基于Python開(kāi)發(fā)簡(jiǎn)易局域網(wǎng)聊天工具的文章就介紹到這了,更多相關(guān)Python局域網(wǎng)聊天內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
python3去掉string中的標(biāo)點(diǎn)符號(hào)方法
今天小編就為大家分享一篇python3去掉string中的標(biāo)點(diǎn)符號(hào)方法，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2019-01-01
django創(chuàng)建自定義模板處理器的實(shí)例詳解
這篇文章主要介紹了django創(chuàng)建自定義模板處理器的實(shí)例詳解的相關(guān)資料,這里說(shuō)明了如何需要django模板處理器及實(shí)現(xiàn)方法，希望大家能理解掌握這部分內(nèi)容，需要的朋友可以參考下2017-08-08
這篇文章主要介紹了python基礎(chǔ)之匿名函數(shù)詳解,文中有非常詳細(xì)的代碼示例,對(duì)正在學(xué)習(xí)python基礎(chǔ)的小伙伴們有很好的幫助,需要的朋友可以參考下2021-04-04
python中的scapy抓取http報(bào)文內(nèi)容
這篇文章主要介紹了python中的scapy抓取http報(bào)文內(nèi)容方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2023-08-08
python 實(shí)現(xiàn)PIL模塊在圖片畫(huà)線寫(xiě)字
這篇文章主要介紹了python 實(shí)現(xiàn)PIL模塊在圖片畫(huà)線寫(xiě)字，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2020-05-05
下面小編就為大家?guī)?lái)一篇Python之re操作方法(詳解)。小編覺(jué)得挺不錯(cuò)的，現(xiàn)在就分享給大家，也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧2017-06-06
python socket多線程實(shí)現(xiàn)客戶端與服務(wù)器連接
這篇文章主要為大家詳細(xì)介紹了python socket多線程實(shí)現(xiàn)客戶端與服務(wù)器連接，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2021-09-09
今天給大家整理了如何使用Python實(shí)現(xiàn)批量處理工作簿和工作表,文中有非常詳細(xì)的介紹及代碼示例,對(duì)小伙伴們很有幫助,需要的朋友可以參考下2021-06-06
簡(jiǎn)單實(shí)例帶你了解Python的編譯和執(zhí)行全過(guò)程
python 是一種解釋型的編程語(yǔ)言，所以不像編譯型語(yǔ)言那樣需要顯式的編譯過(guò)程。然而，在 Python 代碼執(zhí)行之前，它需要被解釋器轉(zhuǎn)換成字節(jié)碼，這個(gè)過(guò)程就是 Python 的編譯過(guò)程,還不知道的朋友快來(lái)看看吧2023-04-04
Python 實(shí)現(xiàn)進(jìn)度條的六種方式
這篇文章主要介紹了Python 實(shí)現(xiàn)進(jìn)度條的六種方式，幫助大家更好的理解和使用python，感興趣的朋友可以了解下2021-01-01