基于Python开发简易局域网聊天工具

更新时间：2025年07月28日 09:02:27 作者：超级小识

端对端加密技术可以确保只有通信的双方能够读取消息内容,即使是服务器也无法解密,本文将详细介绍如何用Python实现一个简单的局域网聊天工具,并为其添加端对端加密功能,感兴趣的小伙伴可以了解下

在现代互联网环境中，隐私和安全越来越受到重视。端对端加密（End-to-End Encryption, E2EE）技术可以确保只有通信的双方能够读取消息内容，即使是服务器也无法解密。本文将详细介绍如何用Python实现一个简单的局域网聊天工具，并为其添加端对端加密功能。

什么是端对端加密

端对端加密是一种通信加密方式，消息在发送端加密后，只有接收端能够解密。中间的任何节点（如路由器、服务器等）都只能看到加密后的数据，无法获取原始消息内容。这种加密方式广泛应用于即时通讯工具中，如WhatsApp、Signal等。

项目概述

本项目将实现以下功能：

基于Python的局域网聊天工具
支持多客户端连接
使用非对称加密（RSA）进行密钥交换
使用对称加密（AES）加密通信内容
简单的命令行界面

技术栈

编程语言：Python 3.x

网络库：socket、threading

加密库：cryptography

其他：argparse（参数解析）

核心模块解析

网络通信基础

局域网聊天工具的核心是网络通信。Python的socket库提供了低级别的网络接口，可以创建TCP或UDP连接。本项目中采用TCP协议，因为它能保证消息的可靠传输。

import socket

# 创建TCP socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

TCP通信需要明确的客户端和服务器端。服务器监听特定端口，客户端主动连接服务器。在聊天工具中，每个用户既是客户端（发送消息）又是服务器（接收消息）。

多线程处理

为了实现同时接收和发送消息，需要使用多线程。Python的threading模块可以方便地创建和管理线程。

import threading

def receive_messages():
    while True:
        data = client_socket.recv(1024)
        print(f"Received: {data.decode()}")

# 创建接收消息的线程
receive_thread = threading.Thread(target=receive_messages)
receive_thread.start()

加密实现

加密部分分为两个阶段：密钥交换和消息加密。

密钥交换：使用RSA非对称加密。每个用户生成自己的RSA密钥对，公开公钥，保存私钥。当两个用户通信时，他们交换公钥，然后用对方的公钥加密一个随机的AES密钥（会话密钥）。

消息加密：使用AES对称加密。一旦会话密钥安全交换，后续通信都使用这个密钥进行加密解密，因为对称加密比非对称加密效率高很多。

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

详细实现步骤

用户类设计

首先设计一个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()
        )

密钥交换协议

设计一个简单的协议来交换公钥和会话密钥：

连接建立后，双方交换RSA公钥
一方生成AES密钥，用对方的公钥加密后发送
对方收到后用私钥解密获取AES密钥
后续通信使用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:
        # 生成并发送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
            )
        )

消息加密解密

实现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()

服务器和客户端实现

将上述功能整合到服务器和客户端代码中：

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()
        
        # 发送消息
        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)

完整源代码

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

在另一台机器上运行客户端（确保在同一局域网）：

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

开始聊天，输入的消息会自动加密传输

安全注意事项

本项目仅用于学习目的，不应用于生产环境
实际应用中需要更完善的密钥管理和身份验证机制
加密实现使用了cryptography库，这是Python中比较可靠的加密库
确保Python环境是最新版本，避免已知的安全漏洞

通过这个项目，您可以学习到网络编程、多线程、加密算法等多项技术。

到此这篇关于基于Python开发简易局域网聊天工具的文章就介绍到这了,更多相关Python局域网聊天内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

python3去掉string中的标点符号方法
今天小编就为大家分享一篇python3去掉string中的标点符号方法，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2019-01-01
django创建自定义模板处理器的实例详解
这篇文章主要介绍了django创建自定义模板处理器的实例详解的相关资料,这里说明了如何需要django模板处理器及实现方法，希望大家能理解掌握这部分内容，需要的朋友可以参考下
2017-08-08
python基础之匿名函数详解
这篇文章主要介绍了python基础之匿名函数详解,文中有非常详细的代码示例,对正在学习python基础的小伙伴们有很好的帮助,需要的朋友可以参考下
2021-04-04
python中的scapy抓取http报文内容
这篇文章主要介绍了python中的scapy抓取http报文内容方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
2023-08-08
python 实现PIL模块在图片画线写字
这篇文章主要介绍了python 实现PIL模块在图片画线写字，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2020-05-05
Python之re操作方法(详解)
下面小编就为大家带来一篇Python之re操作方法(详解)。小编觉得挺不错的，现在就分享给大家，也给大家做个参考。一起跟随小编过来看看吧
2017-06-06
python socket多线程实现客户端与服务器连接
这篇文章主要为大家详细介绍了python socket多线程实现客户端与服务器连接，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2021-09-09
Python自动化之批量处理工作簿和工作表
今天给大家整理了如何使用Python实现批量处理工作簿和工作表,文中有非常详细的介绍及代码示例,对小伙伴们很有帮助,需要的朋友可以参考下
2021-06-06
简单实例带你了解Python的编译和执行全过程
python 是一种解释型的编程语言，所以不像编译型语言那样需要显式的编译过程。然而，在 Python 代码执行之前，它需要被解释器转换成字节码，这个过程就是 Python 的编译过程,还不知道的朋友快来看看吧
2023-04-04
Python 实现进度条的六种方式
这篇文章主要介绍了Python 实现进度条的六种方式，帮助大家更好的理解和使用python，感兴趣的朋友可以了解下
2021-01-01