feat: add SSL/WSS connection support

Timandes · Timandes · commit ba41a8cc29a5 · 2026-02-27T09:44:03.000+08:00
- Add use_ssl parameter to connect() for WSS protocol
- Add skip_ssl_verify parameter to control certificate verification
- Support endpoint with ws:// or wss:// prefix
- Update demo.py with --use-ssl and --skip-ssl-verify CLI args
- Update documentation
- Add spec-driven development docs
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -7,6 +7,12 @@
 
 ## [Unreleased]
 
+### Added
+- `FnosClient.connect()` 新增 SSL/WSS 连接支持
+  - `use_ssl` 参数：是否使用 WSS 协议连接（默认 False）
+  - `skip_ssl_verify` 参数：是否跳过 SSL 证书验证（默认 True，便于自签名证书场景）
+  - 支持 endpoint 带协议前缀（`wss://` 或 `ws://`），前缀优先于 `use_ssl` 参数
+
 ## [0.11.0] - 2026-02-15
 
 ### Added
diff --git a/README.md b/README.md
@@ -30,6 +30,8 @@ async def main():
     parser.add_argument('--user', type=str, required=True, help='用户名')
     parser.add_argument('--password', type=str, required=True, help='密码')
     parser.add_argument('-e', '--endpoint', type=str, default='your-custom-endpoint.com:5666', help='服务器地址 (默认: your-custom-endpoint.com:5666)')
+    parser.add_argument('--use-ssl', action='store_true', help='使用 SSL/WSS 连接')
+    parser.add_argument('--skip-ssl-verify', type=lambda x: x.lower() == 'true', default=True, help='跳过 SSL 证书验证 (默认: True)')
 
     args = parser.parse_args()
 
@@ -39,7 +41,8 @@ async def main():
     client.on_message(on_message_handler)
 
     # 连接到服务器（必须指定endpoint）
-    await client.connect(args.endpoint)
+    # 可选参数：use_ssl=True 使用 WSS 连接，skip_ssl_verify=False 验证证书
+    await client.connect(args.endpoint, use_ssl=args.use_ssl, skip_ssl_verify=args.skip_ssl_verify)
 
     # 登录
     result = await client.login(args.user, args.password)
@@ -54,7 +57,7 @@ async def main():
     # 演示重连功能（手动方式）
     await client.close()  # 先关闭连接
     print("连接已关闭，尝试重连...")
-    await client.connect(args.endpoint)  # 重新连接（现在会等待连接完成）
+    await client.connect(args.endpoint, use_ssl=args.use_ssl, skip_ssl_verify=args.skip_ssl_verify)  # 重新连接（现在会等待连接完成）
     result = await client.login(args.user, args.password)  # 重新登录
     print("重连登录结果:", result)
 
@@ -74,7 +77,7 @@ if __name__ == "__main__":
 | 类名 | 方法名 | 简介 |
 | ---- | ---- | ---- |
 | FnosClient | `__init__` | 初始化客户端，支持type参数（"main"、"timer"或"file"，默认为"main"） |
-| FnosClient | `connect` | 连接到WebSocket服务器（必填参数：endpoint） |
+| FnosClient | `connect` | 连接到WebSocket服务器（必填参数：endpoint；可选参数：use_ssl、skip_ssl_verify） |
 | FnosClient | `login` | 用户登录方法 |
 | FnosClient | `get_decrypted_secret` | 获取解密后的secret |
 | FnosClient | `on_message` | 设置消息回调函数 |
@@ -135,6 +138,9 @@ if __name__ == "__main__":
 - `--user`: 用户名（必填）
 - `--password`: 密码（必填）
 - `-e, --endpoint`: 服务器地址（可选，默认为 your-custom-endpoint.com:5666）
+  - 支持协议前缀：`wss://host:port` 或 `ws://host:port`
+- `--use-ssl`: 使用 SSL/WSS 连接（可选，默认不使用）
+- `--skip-ssl-verify`: 跳过 SSL 证书验证（可选，默认为 True）
 
 ## 运行示例
 
@@ -167,4 +173,19 @@ uv run examples/user.py --user myuser --password mypassword -e my-server.com:566
 | `docker_manager.py` | 演示DockerManager模块的各种功能（Docker Compose项目、容器、统计信息、系统设置） |
 | `event_logger.py` | 演示EventLogger模块的功能（获取事件日志） |
 | `share.py` | 演示Share模块的功能（获取SMB共享配置信息） |
-| `notify.py` | 演示Notify模块的功能（获取未读通知数） |
+| `notify.py` | 演示Notify模块的功能（获取未读通知数） |
+
+### SSL/WSS 连接示例
+
+如果服务器使用 HTTPS/WSS 协议，可以通过以下方式连接：
+
+```bash
+# 方式1：使用 wss:// 前缀
+uv run examples/user.py --user myuser --password mypassword -e wss://my-server.com:5667
+
+# 方式2：使用 --use-ssl 参数
+uv run examples/user.py --user myuser --password mypassword -e my-server.com:5667 --use-ssl
+
+# 禁用证书验证跳过（验证服务器证书）
+uv run examples/user.py --user myuser --password mypassword -e wss://my-server.com:5667 --skip-ssl-verify false
+```
diff --git a/demo.py b/demo.py
@@ -27,6 +27,8 @@ async def main():
     parser.add_argument('--user', type=str, required=True, help='用户名')
     parser.add_argument('--password', type=str, required=True, help='密码')
     parser.add_argument('-e', '--endpoint', type=str, default='your-custom-endpoint.com:5666', help='服务器地址 (默认: your-custom-endpoint.com:5666)')
+    parser.add_argument('--use-ssl', action='store_true', help='使用 SSL/WSS 连接')
+    parser.add_argument('--skip-ssl-verify', type=lambda x: x.lower() == 'true', default=True, help='跳过 SSL 证书验证 (默认: True)')
     
     args = parser.parse_args()
     
@@ -36,7 +38,7 @@ async def main():
     client.on_message(on_message_handler)
     
     # 连接到服务器（必须指定endpoint）
-    await client.connect(args.endpoint)
+    await client.connect(args.endpoint, use_ssl=args.use_ssl, skip_ssl_verify=args.skip_ssl_verify)
     
     if client.connected:
         print("连接成功，尝试登录...")
diff --git a/fnos/client.py b/fnos/client.py
@@ -21,6 +21,7 @@
 import hashlib
 import hmac
 import logging
+import ssl
 from Crypto.PublicKey import RSA
 from Crypto.Cipher import AES, PKCS1_v1_5
 from Crypto.Random import get_random_bytes
@@ -74,6 +75,9 @@ def __init__(self, type: str = "main"):
         self.password = None
         self.token = None
         self.long_token = None
+        # SSL 配置
+        self.use_ssl = False
+        self.skip_ssl_verify = True
 
     def _generate_reqid(self):
         """生成唯一的reqid"""
@@ -92,6 +96,21 @@ def _generate_did(self):
         n = base64.b32encode(str(random.random()).encode()).decode()[:15]
         return f"{t}-{e}-{n}".lower().replace('=', '')
 
+    def _parse_endpoint(self, endpoint: str, use_ssl: bool) -> tuple:
+        """
+        解析 endpoint，返回 (host_port, actual_use_ssl)
+        
+        - 如果 endpoint 以 wss:// 开头，返回去掉前缀的地址和 True
+        - 如果 endpoint 以 ws:// 开头，返回去掉前缀的地址和 False
+        - 否则返回原地址和 use_ssl 参数值
+        """
+        if endpoint.startswith("wss://"):
+            return endpoint[6:], True
+        elif endpoint.startswith("ws://"):
+            return endpoint[5:], False
+        else:
+            return endpoint, use_ssl
+
     def _encrypt_login_data(self, username, password):
         """加密登录数据"""
         # 生成随机AES密钥
@@ -172,14 +191,39 @@ def _decrypt_login_secret(self, encrypted_secret):
             logger.error(f"解密登录secret失败: {e}")
             return None
 
-    async def connect(self, endpoint, timeout: float = 3.0):
-        """连接到WebSocket服务器"""
+    async def connect(self, endpoint, timeout: float = 3.0, use_ssl: bool = False, skip_ssl_verify: bool = True):
+        """连接到WebSocket服务器
+        
+        Args:
+            endpoint: 服务器地址，可以是 "host:port" 格式或带协议前缀 "ws://host:port" / "wss://host:port"
+            timeout: 连接超时时间（秒）
+            use_ssl: 是否使用 SSL/WSS 连接（默认 False）
+            skip_ssl_verify: 是否跳过 SSL 证书验证（默认 True）
+        """
         try:
             logger.info("正在连接到WebSocket服务器...")
-            # 保存endpoint用于重连
-            self.endpoint = endpoint
-            # 创建WebSocket连接，使用构造函数设置的type参数
-            self.ws = await websockets.connect(f"ws://{endpoint}/websocket?type={self.type}")
+            # 解析 endpoint，处理协议前缀
+            parsed_endpoint, actual_use_ssl = self._parse_endpoint(endpoint, use_ssl)
+            
+            # 保存连接信息用于重连
+            self.endpoint = parsed_endpoint
+            self.use_ssl = actual_use_ssl
+            self.skip_ssl_verify = skip_ssl_verify
+            
+            # 根据 use_ssl 选择协议
+            protocol = "wss" if actual_use_ssl else "ws"
+            uri = f"{protocol}://{parsed_endpoint}/websocket?type={self.type}"
+            
+            # 配置 SSL 上下文
+            ssl_context = None
+            if actual_use_ssl:
+                ssl_context = ssl.create_default_context()
+                if skip_ssl_verify:
+                    ssl_context.check_hostname = False
+                    ssl_context.verify_mode = ssl.CERT_NONE
+            
+            # 创建WebSocket连接
+            self.ws = await websockets.connect(uri, ssl=ssl_context)
             logger.debug("websockets.connect returned")
 
             logger.debug("Creating async message handler...")
diff --git a/specs/ssl-support-plan.md b/specs/ssl-support-plan.md
@@ -0,0 +1,103 @@
+# WSS/SSL 连接支持 - 技术实现计划
+
+## 修改范围
+
+### 主要文件
+
+| 文件 | 修改类型 | 说明 |
+|-----|---------|-----|
+| `fnos/client.py` | 修改 | 添加 SSL 支持逻辑 |
+| `demo.py` | 修改 | 添加命令行参数支持 |
+
+## 实现细节
+
+### 1. FnosClient 类变更
+
+#### 1.1 新增实例属性
+
+在 `__init__()` 中添加：
+
+```python
+self.use_ssl = False
+self.skip_ssl_verify = True
+```
+
+#### 1.2 connect() 方法变更
+
+**新增参数：**
+- `use_ssl: bool = False` - 是否使用 SSL 连接
+- `skip_ssl_verify: bool = True` - 是否跳过证书验证
+
+**endpoint 解析逻辑：**
+```python
+def _parse_endpoint(self, endpoint: str, use_ssl: bool) -> tuple[str, bool]:
+    """
+    解析 endpoint，返回 (host_port, actual_use_ssl)
+    
+    - 如果 endpoint 以 wss:// 开头，返回去掉前缀的地址和 True
+    - 如果 endpoint 以 ws:// 开头，返回去掉前缀的地址和 False
+    - 否则返回原地址和 use_ssl 参数值
+    """
+    if endpoint.startswith("wss://"):
+        return endpoint[6:], True
+    elif endpoint.startswith("ws://"):
+        return endpoint[5:], False
+    else:
+        return endpoint, use_ssl
+```
+
+**WebSocket 连接构建：**
+```python
+# 根据 use_ssl 选择协议
+protocol = "wss" if use_ssl else "ws"
+uri = f"{protocol}://{endpoint}/websocket?type={self.type}"
+
+# 配置 SSL 上下文
+ssl_context = None
+if use_ssl and skip_ssl_verify:
+    ssl_context = ssl.create_default_context()
+    ssl_context.check_hostname = False
+    ssl_context.verify_mode = ssl.CERT_NONE
+
+# 创建连接
+self.ws = await websockets.connect(uri, ssl=ssl_context)
+```
+
+#### 1.3 reconnect() 方法变更
+
+无需修改，因为 `connect()` 会自动使用保存的 SSL 配置。
+
+### 2. demo.py 变更
+
+添加命令行参数：
+
+```python
+parser.add_argument('--use-ssl', action='store_true', help='使用 SSL/WSS 连接')
+parser.add_argument('--skip-ssl-verify', type=lambda x: x.lower() == 'true', 
+                    default=True, help='跳过 SSL 证书验证 (默认: True)')
+```
+
+## 依赖
+
+- `ssl` 模块（Python 标准库）
+- `websockets` 库已支持 SSL 参数
+
+## 测试策略
+
+### 单元测试
+
+- 测试 endpoint 解析逻辑（各种前缀组合）
+- 测试协议选择逻辑
+
+### 集成测试
+
+- 测试 ws:// 连接（现有行为）
+- 测试 wss:// 连接（需要测试服务器）
+- 测试自签名证书跳过验证
+
+## 风险与缓解
+
+| 风险 | 缓解措施 |
+|-----|---------|
+| 向后兼容性破坏 | 默认值保持 `use_ssl=False` |
+| SSL 验证失败 | 默认 `skip_ssl_verify=True` 便于自签名证书场景 |
diff --git a/specs/ssl-support-spec.md b/specs/ssl-support-spec.md
@@ -0,0 +1,71 @@
+# WSS/SSL 连接支持规范
+
+## 概述
+
+为 `FnosClient` 添加 HTTPS/WSS (安全 WebSocket) 连接支持，允许用户通过 SSL/TLS 加密连接到服务器。
+
+## 功能需求
+
+### FR-1: use_ssl 参数
+
+**WHEN** 用户调用 `connect()` 方法时，
+**THEN THE SYSTEM SHALL** 接受一个 `use_ssl` 布尔参数（默认 `False`），
+**AND** 当 `use_ssl=True` 时使用 `wss://` 协议连接。
+
+### FR-2: endpoint 协议前缀支持
+
+**WHEN** 用户在 `endpoint` 参数中指定 `ws://` 或 `wss://` 前缀时，
+**THEN THE SYSTEM SHALL** 自动解析并使用该协议，
+**AND** endpoint 中指定的协议前缀优先于 `use_ssl` 参数。
+
+### FR-3: SSL 证书验证控制
+
+**WHEN** 用户连接到使用自签名证书的服务器时，
+**THEN THE SYSTEM SHALL** 提供 `skip_ssl_verify` 参数（默认 `True`）控制是否跳过证书验证，
+**AND** 当 `skip_ssl_verify=True` 时跳过 SSL 证书验证，
+**AND** 当 `skip_ssl_verify=False` 时验证 SSL 证书。
+
+### FR-4: 重连时保持 SSL 配置
+
+**WHEN** 用户调用 `reconnect()` 方法时，
+**THEN THE SYSTEM SHALL** 自动使用之前保存的 SSL 配置（`use_ssl` 和 `skip_ssl_verify`）。
+
+## 参数行为矩阵
+
+| endpoint 格式 | use_ssl | 实际协议 |
+|--------------|---------|---------|
+| `wss://host:port` | 任意 | wss:// |
+| `ws://host:port` | 任意 | ws:// |
+| `host:port` | False | ws:// |
+| `host:port` | True | wss:// |
+
+## API 变更
+
+### connect() 方法签名变更
+
+```python
+async def connect(
+    self, 
+    endpoint, 
+    timeout: float = 3.0,
+    use_ssl: bool = False,
+    skip_ssl_verify: bool = True
+):
+```
+
+### 新增实例属性
+
+- `self.use_ssl: bool` - 保存 SSL 配置用于重连
+- `self.skip_ssl_verify: bool` - 保存证书验证配置用于重连
+
+## 验收标准
+
+- [ ] `use_ssl=True` 时使用 wss:// 协议连接
+- [ ] `use_ssl=False`（默认）时使用 ws:// 协议连接
+- [ ] endpoint 包含 `wss://` 前缀时使用 wss:// 协议
+- [ ] endpoint 包含 `ws://` 前缀时使用 ws:// 协议
+- [ ] endpoint 中的协议前缀优先于 use_ssl 参数
+- [ ] `skip_ssl_verify=True` 时跳过证书验证
+- [ ] `skip_ssl_verify=False` 时验证证书
+- [ ] `reconnect()` 方法使用保存的 SSL 配置
+- [ ] 向后兼容：现有代码无需修改即可正常工作
diff --git a/specs/ssl-support-tasks.md b/specs/ssl-support-tasks.md
