root
4 months ago
6 changed files with 313 additions and 2 deletions
@ -0,0 +1,112 @@ |
|||
# WebSocket SSL证书验证修复说明 |
|||
|
|||
## 问题描述 |
|||
|
|||
在使用WebSocket客户端连接时，遇到SSL证书验证失败的错误： |
|||
|
|||
``` |
|||
[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: self signed certificate (_ssl.c:997) |
|||
``` |
|||
|
|||
## 解决方案 |
|||
|
|||
### 1. 配置化SSL证书验证 |
|||
|
|||
系统现在支持通过配置来控制SSL证书验证行为，而不是硬编码跳过验证。 |
|||
|
|||
### 2. 环境变量配置 |
|||
|
|||
可以通过环境变量来控制SSL验证行为： |
|||
|
|||
```bash |
|||
# 开发环境（跳过SSL验证） |
|||
export WEBSOCKET_SSL_VERIFY_CERTIFICATE=false |
|||
export WEBSOCKET_SSL_VERIFY_HOSTNAME=false |
|||
|
|||
# 生产环境（启用SSL验证） |
|||
export WEBSOCKET_SSL_VERIFY_CERTIFICATE=true |
|||
export WEBSOCKET_SSL_VERIFY_HOSTNAME=true |
|||
``` |
|||
|
|||
### 3. 配置文件设置 |
|||
|
|||
也可以在 `.env` 文件中设置： |
|||
|
|||
```env |
|||
# WebSocket SSL配置 |
|||
WEBSOCKET_SSL_VERIFY_CERTIFICATE=false |
|||
WEBSOCKET_SSL_VERIFY_HOSTNAME=false |
|||
``` |
|||
|
|||
### 4. 默认配置 |
|||
|
|||
- **开发环境**：默认跳过SSL证书验证（`ssl_verify_certificate=false`） |
|||
- **生产环境**：建议启用SSL证书验证（`ssl_verify_certificate=true`） |
|||
|
|||
## 使用方法 |
|||
|
|||
### 1. 安装依赖 |
|||
|
|||
确保已安装websockets依赖： |
|||
|
|||
```bash |
|||
pip install websockets==12.0 |
|||
``` |
|||
|
|||
### 2. 创建WebSocket客户端 |
|||
|
|||
```python |
|||
from app.core.websocket.client import WebSocketClient |
|||
|
|||
# 创建客户端 |
|||
client = WebSocketClient("wss://your-server.com", "my_client") |
|||
|
|||
# 连接（会自动根据配置处理SSL验证） |
|||
success = await client.connect() |
|||
``` |
|||
|
|||
### 3. 测试验证 |
|||
|
|||
运行测试脚本验证修复效果： |
|||
|
|||
```bash |
|||
python test_websocket_ssl.py |
|||
``` |
|||
|
|||
## 安全注意事项 |
|||
|
|||
1. **开发环境**：可以安全地跳过SSL证书验证 |
|||
2. **生产环境**：强烈建议启用SSL证书验证，使用有效的SSL证书 |
|||
3. **自签名证书**：如果使用自签名证书，需要将证书添加到信任链中 |
|||
|
|||
## 相关文件 |
|||
|
|||
- `app/core/websocket/client.py` - WebSocket客户端实现 |
|||
- `app/core/config/settings.py` - 配置管理 |
|||
- `test_websocket_ssl.py` - 测试脚本 |
|||
- `requirements.txt` - 依赖管理 |
|||
|
|||
## 故障排除 |
|||
|
|||
### 1. 仍然出现SSL错误 |
|||
|
|||
检查配置是否正确设置： |
|||
|
|||
```python |
|||
from app.core.config.settings import config |
|||
print(f"SSL验证配置: {config.websocket.ssl_verify_certificate}") |
|||
``` |
|||
|
|||
### 2. 生产环境需要跳过验证 |
|||
|
|||
如果生产环境必须跳过SSL验证，可以设置环境变量： |
|||
|
|||
```bash |
|||
export WEBSOCKET_SSL_VERIFY_CERTIFICATE=false |
|||
``` |
|||
|
|||
### 3. 使用自签名证书 |
|||
|
|||
如果使用自签名证书，建议： |
|||
1. 将证书添加到系统信任链 |
|||
2. 或者使用配置跳过验证（仅限开发环境） |
|||
@ -0,0 +1,99 @@ |
|||
#!/usr/bin/env python3 |
|||
""" |
|||
WebSocket SSL证书验证测试脚本 |
|||
""" |
|||
import asyncio |
|||
import sys |
|||
import os |
|||
|
|||
# 添加项目根目录到Python路径 |
|||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) |
|||
|
|||
from app.core.websocket.client import WebSocketClient |
|||
from app.core.config.settings import config |
|||
from app.utils.log import get_logger |
|||
|
|||
logger = get_logger(__name__) |
|||
|
|||
async def test_websocket_ssl(): |
|||
"""测试WebSocket SSL连接""" |
|||
print("=== WebSocket SSL证书验证测试 ===") |
|||
|
|||
# 显示当前配置 |
|||
print(f"SSL证书验证: {config.websocket.ssl_verify_certificate}") |
|||
print(f"SSL主机名验证: {config.websocket.ssl_verify_hostname}") |
|||
|
|||
# 测试URL（使用自签名证书的WebSocket服务器） |
|||
test_url = "wss://echo.websocket.org" # 使用公共测试服务器 |
|||
|
|||
print(f"\n测试连接到: {test_url}") |
|||
|
|||
try: |
|||
# 创建WebSocket客户端 |
|||
client = WebSocketClient(test_url, "test_client") |
|||
|
|||
# 尝试连接 |
|||
print("正在连接...") |
|||
success = await client.connect() |
|||
|
|||
if success: |
|||
print("✅ 连接成功！SSL证书验证问题已修复") |
|||
|
|||
# 发送测试消息 |
|||
test_message = { |
|||
"type": "test", |
|||
"data": "Hello WebSocket!" |
|||
} |
|||
|
|||
# 注册消息处理器 |
|||
def handle_response(message): |
|||
print(f"收到响应: {message}") |
|||
|
|||
client.register_message_handler("test", handle_response) |
|||
|
|||
# 等待一段时间接收响应 |
|||
await asyncio.sleep(2) |
|||
|
|||
# 断开连接 |
|||
await client.disconnect() |
|||
print("连接已断开") |
|||
else: |
|||
print("❌ 连接失败") |
|||
|
|||
except Exception as e: |
|||
print(f"❌ 测试过程中出现异常: {e}") |
|||
logger.error(f"WebSocket SSL测试失败: {e}") |
|||
|
|||
async def test_config_override(): |
|||
"""测试配置覆盖功能""" |
|||
print("\n=== 配置覆盖测试 ===") |
|||
|
|||
# 临时修改配置 |
|||
original_verify_cert = config.websocket.ssl_verify_certificate |
|||
original_verify_host = config.websocket.ssl_verify_hostname |
|||
|
|||
try: |
|||
# 设置为启用验证（应该会失败） |
|||
config.websocket.ssl_verify_certificate = True |
|||
config.websocket.ssl_verify_hostname = True |
|||
|
|||
print("配置已设置为启用SSL验证") |
|||
print(f"SSL证书验证: {config.websocket.ssl_verify_certificate}") |
|||
print(f"SSL主机名验证: {config.websocket.ssl_verify_hostname}") |
|||
|
|||
# 这里可以添加更多测试逻辑 |
|||
|
|||
finally: |
|||
# 恢复原始配置 |
|||
config.websocket.ssl_verify_certificate = original_verify_cert |
|||
config.websocket.ssl_verify_hostname = original_verify_host |
|||
print("配置已恢复") |
|||
|
|||
if __name__ == "__main__": |
|||
print("开始WebSocket SSL测试...") |
|||
|
|||
# 运行测试 |
|||
asyncio.run(test_websocket_ssl()) |
|||
asyncio.run(test_config_override()) |
|||
|
|||
print("\n测试完成！") |
|||
Loading…
Reference in new issue