OTA接口授权与认证

main/xiaozhi-server/config.yaml

@@ -31,6 +31,8 @@ server:
   auth:
     # 是否启用认证
     enabled: false
+    # token签名秘钥
+    signature_key: "your-token-signature-key"
     # 设备的token，可以在编译固件的环节，写入你自己定义的token
     # 固件上的token和以下的token如果能对应，才能连接本服务端
     tokens:

main/xiaozhi-server/core/api/ota_handler.py

@@ -4,6 +4,8 @@ import base64
 import hashlib
 import hmac
 from aiohttp import web
+
+from core.ota_auth import AuthManager
 from core.utils.util import get_local_ip
 from core.api.base_handler import BaseHandler
 
@@ -13,6 +15,15 @@ TAG = __name__
 class OTAHandler(BaseHandler):
     def __init__(self, config: dict):
         super().__init__(config)
+        auth_config = config["server"].get("auth", {})
+        self.auth_enable = auth_config.get("enabled", False)
+        # 设备白名单
+        self.allowed_devices = set(
+            auth_config.get("allowed_devices", [])
+        )
+        secret_key = auth_config.get("signature_key", "")
+        expire_seconds = auth_config.get("expire_seconds")
+        self.auth = AuthManager(secret_key=secret_key, expire_seconds=expire_seconds)
         
     def generate_password_signature(self, content: str, secret_key: str) -> str:
         """生成MQTT密码签名
@@ -64,6 +75,12 @@ class OTAHandler(BaseHandler):
             else:
                 raise Exception("OTA请求设备ID为空")
 
+            client_id = request.headers.get("client-id", "")
+            if client_id:
+                self.logger.bind(tag=TAG).info(f"OTA请求ClientID: {client_id}")
+            else:
+                raise Exception("OTA请求ClientID为空")
+
             data_json = json.loads(data)
 
             server_config = self.config["server"]
@@ -134,10 +151,20 @@ class OTAHandler(BaseHandler):
                 
                 
             else:  # 未配置 mqtt_gateway，下发 WebSocket
+                # 如果开启了认证，则进行认证校验
+                token = ""
+                if self.auth_enable:
+                    if self.allowed_devices:
+                        if device_id in self.allowed_devices:
+                            token = self.auth.generate_token(client_id, device_id)
+                    else:
+                        token = self.auth.generate_token(client_id, device_id)
                 return_json["websocket"] = {
                     "url": self._get_websocket_url(local_ip, port),
+                    "token": token
                 }
                 self.logger.bind(tag=TAG).info(f"未配置MQTT网关，为设备 {device_id} 下发WebSocket配置")
+                self.logger.bind(tag=TAG).info(f"{return_json}")
             
             response = web.Response(
                 text=json.dumps(return_json, separators=(",", ":")),

main/xiaozhi-server/core/connection.py

@@ -10,6 +10,8 @@ import threading
 import traceback
 import subprocess
 import websockets
+
+from core.ota_auth import AuthManager
 from core.utils.util import (
     extract_json_from_string,
     check_vad_update,
@@ -67,7 +69,17 @@ class ConnectionHandler:
         self.logger = setup_logging()
         self.server = server  # 保存server实例的引用
 
-        self.auth = AuthMiddleware(config)
+        # 认证相关
+        auth_config = self.config["server"].get("auth", {})
+        self.auth_enable = auth_config.get("enabled", False)
+        # 设备白名单
+        self.allowed_devices = set(
+            auth_config.get("allowed_devices", [])
+        )
+        secret_key = auth_config.get("signature_key", "")
+        expire_seconds = auth_config.get("expire_seconds", None)
+        self.auth = AuthManager(secret_key=secret_key, expire_seconds=expire_seconds)
+
         self.need_bind = False
         self.bind_code = None
         self.read_config_from_api = self.config.get("read_config_from_api", False)
@@ -194,12 +206,27 @@ class ConnectionHandler:
                 f"{self.client_ip} conn - Headers: {self.headers}"
             )
 
-            # 进行认证
-            await self.auth.authenticate(self.headers)
+            self.device_id = self.headers.get("device-id", None)
+
+            # 是否开启了认证，如果开启了则进行认证,否则不认证
+            if self.auth_enable:
+                # 如果启用了白名单则优先处理白名单
+                if self.allowed_devices and self.device_id not in self.allowed_devices:
+                    raise AuthenticationError("未经授权的DeviceID")
+                else:
+                    # 否则校验token
+                    token = self.headers.get("authorization", "")
+                    if token.startswith('Bearer '):
+                        token = token[7:]  # 移除'Bearer '前缀
+                    else:
+                        raise AuthenticationError("Missing or invalid Authorization header")
+                    # 进行认证
+                    auth_success = self.auth.verify_token(token, client_id=self.headers.get("client-id"), username=self.headers.get("device-id"))
+                    if not auth_success:
+                        raise AuthenticationError("Invalid token")
 
             # 认证通过,继续处理
             self.websocket = ws
-            self.device_id = self.headers.get("device-id", None)
 
             # 检查是否来自MQTT连接
             request_path = ws.request.path

main/xiaozhi-server/core/ota_auth.py

@@ -0,0 +1,64 @@
+import hmac
+import base64
+import hashlib
+import time
+
+
+class AuthManager:
+    """
+    统一授权认证管理器
+    生成与验证 client_id device_id token（HMAC-SHA256）认证三元组
+    token 中不含明文 client_id/device_id，只携带签名 + 时间戳; client_id/device_id在连接时传递
+    在 MQTT 中 client_id: client_id, username: device_id, password: token
+    在 Websocket 中，header:{Device-ID: device_id, Client-ID: client_id, Authorization: Bearer token, ......}
+    """
+
+    def __init__(self, secret_key: str, expire_seconds: int = 60 * 60 * 24 * 30):
+        if not expire_seconds or expire_seconds < 0:
+            self.expire_seconds = 60 * 60 * 24 * 30
+        else:
+            self.expire_seconds = expire_seconds
+        self.secret_key = secret_key
+
+    def _sign(self, content: str) -> str:
+        """HMAC-SHA256签名并Base64编码"""
+        sig = hmac.new(self.secret_key.encode("utf-8"), content.encode("utf-8"), hashlib.sha256).digest()
+        return base64.urlsafe_b64encode(sig).decode("utf-8").rstrip("=")
+
+    def generate_token(self, client_id: str, username: str) -> str:
+        """
+        生成 token
+        Args:
+            client_id: 设备连接ID
+            username: 设备用户名（通常为deviceId）
+        Returns:
+            str: token字符串
+        """
+        ts = int(time.time())
+        content = f"{client_id}|{username}|{ts}"
+        signature = self._sign(content)
+        # token仅包含签名与时间戳，不包含明文信息
+        token = f"{signature}.{ts}"
+        return token
+
+    def verify_token(self, token: str, client_id: str, username: str) -> bool:
+        """
+        验证token有效性
+        Args:
+            token: 客户端传入的token
+            client_id: 连接使用的client_id
+            username: 连接使用的username
+        """
+        try:
+            sig_part, ts_str = token.split(".")
+            ts = int(ts_str)
+            if int(time.time()) - ts > self.expire_seconds:
+                return False  # 过期
+
+            expected_sig = self._sign(f"{client_id}|{username}|{ts}")
+            if not hmac.compare_digest(sig_part, expected_sig):
+                return False
+
+            return True
+        except Exception:
+            return False