diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml
index 4acae788..261f68cf 100644
--- a/main/xiaozhi-server/config.yaml
+++ b/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:
diff --git a/main/xiaozhi-server/core/api/ota_handler.py b/main/xiaozhi-server/core/api/ota_handler.py
index b93a7ed3..44eee43b 100644
--- a/main/xiaozhi-server/core/api/ota_handler.py
+++ b/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=(",", ":")),
diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py
index 8db70fbf..bd14a6fd 100644
--- a/main/xiaozhi-server/core/connection.py
+++ b/main/xiaozhi-server/core/connection.py
@@ -10,6 +10,7 @@ import threading
 import traceback
 import subprocess
 import websockets
+
 from core.utils.util import (
     extract_json_from_string,
     check_vad_update,
@@ -67,7 +68,6 @@ class ConnectionHandler:
         self.logger = setup_logging()
         self.server = server  # 保存server实例的引用
 
-        self.auth = AuthMiddleware(config)
         self.need_bind = False
         self.bind_code = None
         self.read_config_from_api = self.config.get("read_config_from_api", False)
@@ -194,12 +194,10 @@ 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)
 
             # 认证通过,继续处理
             self.websocket = ws
-            self.device_id = self.headers.get("device-id", None)
 
             # 检查是否来自MQTT连接
             request_path = ws.request.path
diff --git a/main/xiaozhi-server/core/ota_auth.py b/main/xiaozhi-server/core/ota_auth.py
new file mode 100644
index 00000000..3c5e7d9d
--- /dev/null
+++ b/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
diff --git a/main/xiaozhi-server/core/websocket_server.py b/main/xiaozhi-server/core/websocket_server.py
index 09ba7961..b3a5a7d9 100644
--- a/main/xiaozhi-server/core/websocket_server.py
+++ b/main/xiaozhi-server/core/websocket_server.py
@@ -1,8 +1,12 @@
 import asyncio
+import json
+
 import websockets
 from config.logger import setup_logging
+from core.auth import AuthenticationError
 from core.connection import ConnectionHandler
 from config.config_loader import get_config_from_api
+from core.ota_auth import AuthManager
 from core.utils.modules_initialize import initialize_modules
 from core.utils.util import check_vad_update, check_asr_update
 
@@ -32,6 +36,16 @@ class WebSocketServer:
 
         self.active_connections = set()
 
+        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)
+
     async def start(self):
         server_config = self.config["server"]
         host = server_config.get("ip", "0.0.0.0")
@@ -44,6 +58,18 @@ class WebSocketServer:
 
     async def _handle_connection(self, websocket):
         """处理新连接，每次创建独立的ConnectionHandler"""
+        # 先认证，后建立连接
+        try:
+            await self._handle_auth(websocket)
+        except AuthenticationError:
+            await websocket.send(json.dumps({
+                "type": "alert",
+                "status": "ERROR",
+                "message": "认证失败",
+                "emotion": "sad"
+            }))
+            await websocket.close()
+            return
         # 创建ConnectionHandler时传入当前server实例
         handler = ConnectionHandler(
             self.config,
@@ -136,3 +162,24 @@ class WebSocketServer:
         except Exception as e:
             self.logger.bind(tag=TAG).error(f"更新服务器配置失败: {str(e)}")
             return False
+
+    async def _handle_auth(self, websocket):
+        # 先认证，后建立连接
+        if self.auth_enable:
+            headers = dict(websocket.request.headers)
+            device_id = headers.get("device-id", None)
+            client_id = headers.get("client-id", None)
+            # 如果启用了白名单则优先处理白名单
+            if self.allowed_devices and device_id not in self.allowed_devices:
+                raise AuthenticationError("未经授权的DeviceID")
+            else:
+                # 否则校验token
+                token = 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=client_id, username=device_id)
+                if not auth_success:
+                    raise AuthenticationError("Invalid token")