本篇文档中的样例程序是一个简单的 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:59