从零实现wss通信示例（WebSocket SSL）_websocket wss 证书

客户端和服务端代码框架跟上一篇一致，仅增加了ssl的证书部分用于加密通信，明文通信（ws协议）见上一篇【https://blog.csdn.net/suoxd123/article/details/148093934】

1. 证书创建

1. 安装openssl

【官网地址】：https://slproweb.com/products/Win32OpenSSL.html

1.2 生成证书

openssl req -x509 -newkey rsa:4096 -keyout key.pem -out cert.pem -days 365

证书构建参数详解

• -x509 指定生成X.509自签名证书。此参数会跳过生成证书请求（CSR）的步骤，直接输出完整证书。
• -newkey rsa:4096生成一个4096位的RSA私钥。rsa:4096表示密钥长度，安全性更高（默认为2048位）。
• -keyout key.pem指定私钥文件的保存路径为key.pem。私钥用于后续加密和证书签名。
• -out cert.pem指定输出证书的文件名为cert.pem。
• -days 365设置证书有效期为365天。可根据需求调整（如-days 730表示2年）。
• 其他可选参数
  • -nodes：跳过私钥加密（生成无密码的私钥）。
  • -subj “/CN=example.com”：直接指定证书主题名称（Common Name），避免交互式输入

X.509 ：定义了证书的标准化格式，是 TLS/SSL 安全通信的基础，用于快速生成自签名证书，跳过 CSR 和 CA 签名步骤。若不使用 -x509，默认会生成 CSR 文件（需后续由 CA 签名生成正式证书【https://letsencrypt.org/docs/glossary/】）

1.3 查看证书

openssl x509 -in cert.pem -text -noout

2. 服务端代码

import asyncioimport websocketsimport sslasync def echo(ws): async for message in ws: print(f\"Received message: {message}\") await ws.send(f\"你好，收到你的消息: {message}\")ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)ssl_context.load_cert_chain(certfile=\"cert.pem\", keyfile=\"key.pem\")async def main(): print(\"WebSocket服务器启动，等待客户端连接...\") async with websockets.serve(echo, \"localhost\", 12345,ssl=ssl_context): await asyncio.Future() # 保持服务器运行asyncio.run(main())

3. 客户端代码（拷贝证书）

使用服务器生成的证书文件，创建证书时候，common name使用客户端同一个域名（localhost）。

import asyncioimport websocketsimport sslasync def hello(): uri = \"wss://localhost:12345\" ssl_context = ssl.create_default_context() ssl_context.load_verify_locations(cafile=\"cert.pem\") async with websockets.connect(uri, ssl=ssl_context) as ws: msg = \"你好！\" await ws.send(msg) print(f\"发送成功： {msg}\") greeting = await ws.recv() print(f\"收到响应： {greeting}\")asyncio.run(hello())

4. 客户端代码（不验证证书）

import asyncioimport websocketsimport sslasync def hello(): uri = \"wss://localhost:12345\" ssl_context = ssl._create_unverified_context() # 创建一个不验证证书的 SSL 上下文 async with websockets.connect(uri, ssl=ssl_context) as ws: msg = \"你好！\" await ws.send(msg) print(f\"发送成功： {msg}\") greeting = await ws.recv() print(f\"收到响应： {greeting}\")asyncio.run(hello())

5. 客户端代码（不验证主机地址）

import asyncioimport websocketsimport sslasync def hello(): uri = \"wss://127.0.0.1:12345\" ssl_context = ssl.create_default_context() ssl_context.load_verify_locations(cafile=\"cert.pem\") ssl_context.check_hostname = False # 禁用主机名验证 async with websockets.connect(uri, ssl=ssl_context) as ws: msg = \"你好！\" await ws.send(msg) print(f\"发送成功： {msg}\") greeting = await ws.recv() print(f\"收到响应： {greeting}\")asyncio.run(hello())

5. 测试效果