本篇文档中的样例程序是一个简单的 WebSocket SSL/TLS 服务器,它使用了 websockets 库和自签名证书(server.crt 和 server.key)来启用加密通信。

准备环境

1.查看是否已安装 Python 3

python3 --version

如果没有安装,使用以下命令安装 Python 3 和 pip(Python 包管理器)

sudo apt update
sudo apt install python3 python3-pip

安装完后,验证 Python 版本

python3 --version

一.创建自签名证书(如果你还没有证书)

如果你没有现成的证书和私钥文件,可以使用 OpenSSL 来创建自签名证书:
我们将使用 OpenSSL 工具来生成新的证书。如果你之前没有安装 OpenSSL,可以通过以下命令安装:

sudo apt-get install openssl
openssl req -new -newkey rsa:2048 -days 365 -nodes -x509 -keyout server.key -out server.crt

当你在生成证书签名请求(CSR)时,会被要求填写一些信息,这些信息将包含在证书请求中。以下是每个字段的解释:

Country Name (2 letter code) [AU]:

该字段要求输入你所在国家的两字母代码(例如:CN 代表中国,US 代表美国,IN 代表印度等)。
示例:CN(中国)。

State or Province Name (full name) [Some-State]:

这个字段要求输入你所在的省份或州的全名(例如:California、Beijing 等)。
示例:Beijing(北京)。

Locality Name (eg, city) []:

这个字段要求输入你所在城市的名称(例如:Shanghai、New York 等)。
示例:Shanghai(上海)。

Organization Name (eg, company) [Internet Widgits Pty Ltd]:

该字段要求输入你公司的名称或组织的名称。
示例:MyCompany(我的公司)。

Organizational Unit Name (eg, section) []:

这个字段是可选的,用来指定公司内的部门或单位。如果不需要,可以留空。
示例:IT Department(IT 部门)或直接留空。

Common Name (e.g. server FQDN or YOUR name) []:

这个字段非常重要,尤其是在申请 SSL 证书时。这里填写你想要申请证书的域名或子域名。
示例:example.com 或 www.example.com。 我这里本机ip是192.168.1.206.所以写成192.168.1.206

Email Address []:

这是一个可选字段,用于提供联系管理员的电子邮件地址。
示例:admin@example.com
A challenge password []:

这个字段是可选的,如果你不需要设置密码来保护 CSR 文件,可以直接跳过。
An optional company name []:

这个字段也是可选的,通常用于填写公司名称的别名。如果不需要,可以留空。

这将在当前目录下生成以下文件:

  • server.key: 私钥文件,用于加密通信
  • server.crt: 公钥证书文件,用于验证通信方的身份

二. 编写服务端的代码及所需内容

在刚才生成证书的目录下,创建server.py文件,命令为vim server.py
服务端程序内容为:

import asyncio
import websockets
import ssl

async def echo(websocket, path):
    async for message in websocket:
        print(f"Received message: {message}")
        await websocket.send(f"Echo: {message}")

async def main():
    # 创建 SSL 上下文并加载证书和私钥
    ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    ssl_context.load_cert_chain(certfile="./server.crt", keyfile="./server.key")

    # 启动 WebSocket 服务器,监听并启用 SSL
    server = await websockets.serve(
        echo, "127.0.0.1", 8000, ssl=ssl_context
    )

    print("WebSocket server started on wss://127.0.0.1:8000")
    await server.wait_closed()

# 启动事件循环并运行主函数
asyncio.run(main())

运行服务端程序python3 server.py,根据提示,并安装所需的模块,我这里需要安装下面内容

pip install websockets -i https://pypi.tuna.tsinghua.edu.cn/simple

最后,正常运行服务端程序应该如下图所示:

三 .编写客户端的代码及所需内容

再打开一个终端窗口,同样在生成证书的目录下,创建client.py文件,命令为vim client.py
客户端程序内容为:

import asyncio
import websockets
import ssl

async def hello():
    uri = "wss://127.0.0.1:8000"    

    # 创建 SSL 上下文并加载证书和私钥
    ssl_context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
    ssl_context.load_cert_chain(certfile="./server.crt", keyfile="./server.key")

    # 由于是自签名证书,可以选择禁用验证
    ssl_context.check_hostname = False  # 禁用主机名验证
    ssl_context.verify_mode = ssl.CERT_NONE  # 禁用证书验证(仅适用于自签名证书)

    async with websockets.connect(uri, ssl=ssl_context) as websocket:
        message = "Hello, WebSocket!"
        await websocket.send(message)
        print(f"Sent: {message}")

        response = await websocket.recv()
        print(f"Received: {response}")

asyncio.run(hello())

同样,运行命令python3 client.py启动客户端,可以看到客户端结果如下图内容所示:

此时,我们再切换到服务端界面,可以看到:

出来了内容Received message: Hello, WebSocket!,连接成功。

注:如果是不同电脑之间,则需要先在同一个把url这里换成实际的ip地址即可。

作者:zuoyingying  创建时间:2025-02-12 17:20
最后编辑:zuoyingying  更新时间:2025-02-12 17:59