基于Python開發(fā)簡易局域網(wǎng)聊天工具

更新時間：2025年07月28日 09:02:27 作者：超級小識

端對端加密技術可以確保只有通信的雙方能夠讀取消息內(nèi)容,即使是服務器也無法解密,本文將詳細介紹如何用Python實現(xiàn)一個簡單的局域網(wǎng)聊天工具,并為其添加端對端加密功能,感興趣的小伙伴可以了解下

在現(xiàn)代互聯(lián)網(wǎng)環(huán)境中，隱私和安全越來越受到重視。端對端加密（End-to-End Encryption, E2EE）技術可以確保只有通信的雙方能夠讀取消息內(nèi)容，即使是服務器也無法解密。本文將詳細介紹如何用Python實現(xiàn)一個簡單的局域網(wǎng)聊天工具，并為其添加端對端加密功能。

什么是端對端加密

端對端加密是一種通信加密方式，消息在發(fā)送端加密后，只有接收端能夠解密。中間的任何節(jié)點（如路由器、服務器等）都只能看到加密后的數(shù)據(jù)，無法獲取原始消息內(nèi)容。這種加密方式廣泛應用于即時通訊工具中，如WhatsApp、Signal等。

項目概述

本項目將實現(xiàn)以下功能：

基于Python的局域網(wǎng)聊天工具
支持多客戶端連接
使用非對稱加密（RSA）進行密鑰交換
使用對稱加密（AES）加密通信內(nèi)容
簡單的命令行界面

技術棧

編程語言：Python 3.x

網(wǎng)絡庫：socket、threading

加密庫：cryptography

其他：argparse（參數(shù)解析）

核心模塊解析

網(wǎng)絡通信基礎

局域網(wǎng)聊天工具的核心是網(wǎng)絡通信。Python的socket庫提供了低級別的網(wǎng)絡接口，可以創(chuàng)建TCP或UDP連接。本項目中采用TCP協(xié)議，因為它能保證消息的可靠傳輸。

import socket

# 創(chuàng)建TCP socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

TCP通信需要明確的客戶端和服務器端。服務器監(jiān)聽特定端口，客戶端主動連接服務器。在聊天工具中，每個用戶既是客戶端（發(fā)送消息）又是服務器（接收消息）。

多線程處理

為了實現(xiàn)同時接收和發(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()

加密實現(xiàn)

加密部分分為兩個階段：密鑰交換和消息加密。

密鑰交換：使用RSA非對稱加密。每個用戶生成自己的RSA密鑰對，公開公鑰，保存私鑰。當兩個用戶通信時，他們交換公鑰，然后用對方的公鑰加密一個隨機的AES密鑰（會話密鑰）。

消息加密：使用AES對稱加密。一旦會話密鑰安全交換，后續(xù)通信都使用這個密鑰進行加密解密，因為對稱加密比非對稱加密效率高很多。

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密鑰對
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

詳細實現(xiàn)步驟

用戶類設計

首先設計一個User類，包含用戶的基本信息和加密相關操作。

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é)議

設計一個簡單的協(xié)議來交換公鑰和會話密鑰：

連接建立后，雙方交換RSA公鑰
一方生成AES密鑰，用對方的公鑰加密后發(fā)送
對方收到后用私鑰解密獲取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
            )
        )

消息加密解密

實現(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()

服務器和客戶端實現(xiàn)

將上述功能整合到服務器和客戶端代碼中：

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)
        
        # 啟動接收消息線程
        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)
        
        # 啟動接收消息線程
        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")

使用說明

在一臺機器上運行服務器：

python chat.py --server --port 12345

在另一臺機器上運行客戶端（確保在同一局域網(wǎng)）：

python chat.py --client --host <服務器IP> --port 12345

開始聊天，輸入的消息會自動加密傳輸

安全注意事項

本項目僅用于學習目的，不應用于生產(chǎn)環(huán)境
實際應用中需要更完善的密鑰管理和身份驗證機制
加密實現(xiàn)使用了cryptography庫，這是Python中比較可靠的加密庫
確保Python環(huán)境是最新版本，避免已知的安全漏洞

通過這個項目，您可以學習到網(wǎng)絡編程、多線程、加密算法等多項技術。

到此這篇關于基于Python開發(fā)簡易局域網(wǎng)聊天工具的文章就介紹到這了,更多相關Python局域網(wǎng)聊天內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:

Python中實現(xiàn)帶權重的隨機測試
本文主要介紹了Python中實現(xiàn)帶權重的隨機測試,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧
2026-05-05
python scrapy簡單模擬登錄的代碼分析
在本篇文章里小編給大家整理的是一篇關于python scrapy簡單模擬登錄的代碼分析，有興趣的朋友們可以學習參考下。
2021-07-07
在Python的web框架中編寫創(chuàng)建日志的程序的教程
這篇文章主要介紹了在Python的web框架中編寫創(chuàng)建日志的程序的教程,示例代碼基于Python2.x版本,需要的朋友可以參考下
2015-04-04
沒編程基礎可以學python嗎
在本篇文章里小編給大家整理的是關于沒編程基礎可以學python嗎的相關知識點，需要的朋友們可以學習下。
2020-06-06
Pandas中八個常用option設置的示例詳解
通過pandas的使用，我們經(jīng)常要交互式地展示表格（dataframe）、分析表格。本文整理了8個常用的配置選項，使用可以提高很多效率，需要的可以參考一下
2022-06-06
python?opencv實現(xiàn)灰度圖和彩色圖的互相轉換
這篇文章主要給大家介紹了關于python?opencv實現(xiàn)灰度圖和彩色圖的互相轉換,文中通過實例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下
2022-01-01
切換路徑在Jupyter里調(diào)用本地文件的操作
這篇文章主要介紹了切換路徑在Jupyter里調(diào)用本地文件的操作，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧
2021-03-03
Pandas執(zhí)行SQL操作的實現(xiàn)
使用SQL語句能夠完成對table的增刪改查操作,Pandas同樣也可以實現(xiàn)SQL語句的基本功能,本文就來介紹一下,具有一檔的參考價值,感興趣的可以了解一下
2024-07-07
pytorch numpy list類型之間的相互轉換實例
今天小編就為大家分享一篇pytorch numpy list類型之間的相互轉換實例，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧
2019-08-08
將FileStorage對象高效轉換為NumPy數(shù)組的兩種實現(xiàn)方案
在Web開發(fā)（如Flask應用）中,處理用戶上傳的圖片文件時,常會遇到FileStorage對象向numpy.ndarray的轉換需求,本文將提供兩種經(jīng)過驗證的高效方法,并深入解析其技術細節(jié)與適用場景,需要的朋友可以參考下
2025-03-03