diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml
index 4acae788..b2cc9fe6 100644
--- a/main/xiaozhi-server/config.yaml
+++ b/main/xiaozhi-server/config.yaml
@@ -31,16 +31,10 @@ server:
   auth:
     # 是否启用认证
     enabled: false
-    # 设备的token，可以在编译固件的环节，写入你自己定义的token
-    # 固件上的token和以下的token如果能对应，才能连接本服务端
-    tokens:
-      - token: "your-token1" # 设备1的token
-        name: "your-device-name1"  # 设备1标识
-      - token: "your-token2"  # 设备2的token
-        name: "your-device-name2" # 设备2标识
-    # 可选:设备白名单，如果设置了白名单，那么白名单的机器无论是什么token都可以连接。
-    #allowed_devices:
-    #  - "24:0A:C4:1D:3B:F0"  # MAC地址列表
+    # 白名单设备ID列表
+    # 如果属于白名单内的设备，不校验token，直接放行
+    allowed_devices:
+      - "11:22:33:44:55:66"
  # MQTT网关配置，用于通过OTA下发到设备，根据mqtt_gateway的.env文件配置，格式为host:port
   mqtt_gateway: null
   # MQTT签名密钥，用于生成MQTT连接密码，根据mqtt_gateway的.env文件配置
diff --git a/main/xiaozhi-server/core/api/ota_handler.py b/main/xiaozhi-server/core/api/ota_handler.py
index b93a7ed3..d3d37348 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.auth import AuthManager
 from core.utils.util import get_local_ip
 from core.api.base_handler import BaseHandler
 
@@ -13,21 +15,30 @@ 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 = config["server"]["auth_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密码签名
-        
+
         Args:
             content: 签名内容 (clientId + '|' + username)
             secret_key: 密钥
-            
+
         Returns:
             str: Base64编码的HMAC-SHA256签名
         """
         try:
-            hmac_obj = hmac.new(secret_key.encode('utf-8'), content.encode('utf-8'), hashlib.sha256)
+            hmac_obj = hmac.new(
+                secret_key.encode("utf-8"), content.encode("utf-8"), hashlib.sha256
+            )
             signature = hmac_obj.digest()
-            return base64.b64encode(signature).decode('utf-8')
+            return base64.b64encode(signature).decode("utf-8")
         except Exception as e:
             self.logger.bind(tag=TAG).error(f"生成MQTT密码签名失败: {e}")
             return ""
@@ -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"]
@@ -80,10 +97,9 @@ class OTAHandler(BaseHandler):
                     "url": "",
                 },
             }
-            
 
             mqtt_gateway_endpoint = server_config.get("mqtt_gateway")
-            
+
             if mqtt_gateway_endpoint:  # 如果配置了非空字符串
                 # 尝试从请求数据中获取设备型号
                 device_model = "default"
@@ -101,12 +117,12 @@ class OTAHandler(BaseHandler):
                 mqtt_client_id = f"{group_id}@@@{mac_address_safe}@@@{mac_address_safe}"
 
                 # 构建用户数据
-                user_data = {
-                    "ip": "unknown"
-                }
+                user_data = {"ip": "unknown"}
                 try:
                     user_data_json = json.dumps(user_data)
-                    username = base64.b64encode(user_data_json.encode('utf-8')).decode('utf-8')
+                    username = base64.b64encode(user_data_json.encode("utf-8")).decode(
+                        "utf-8"
+                    )
                 except Exception as e:
                     self.logger.bind(tag=TAG).error(f"生成用户名失败: {e}")
                     username = ""
@@ -115,7 +131,9 @@ class OTAHandler(BaseHandler):
                 password = ""
                 signature_key = server_config.get("mqtt_signature_key", "")
                 if signature_key:
-                    password = self.generate_password_signature(mqtt_client_id + "|" + username, signature_key)
+                    password = self.generate_password_signature(
+                        mqtt_client_id + "|" + username, signature_key
+                    )
                     if not password:
                         password = ""  # 签名失败则留空，由设备决定是否允许无密码
                 else:
@@ -128,17 +146,28 @@ class OTAHandler(BaseHandler):
                     "username": username,
                     "password": password,
                     "publish_topic": "device-server",
-                    "subscribe_topic": f"devices/p2p/{mac_address_safe}"
+                    "subscribe_topic": f"devices/p2p/{mac_address_safe}",
                 }
                 self.logger.bind(tag=TAG).info(f"为设备 {device_id} 下发MQTT网关配置")
-                
-                
+
             else:  # 未配置 mqtt_gateway，下发 WebSocket
+                # 如果开启了认证，则进行认证校验
+                token = ""
+                if self.auth_enable:
+                    if self.allowed_devices:
+                        if device_id not 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"未配置MQTT网关，为设备 {device_id} 下发WebSocket配置"
+                )
+                self.logger.bind(tag=TAG).info(f"{return_json}")
+
             response = web.Response(
                 text=json.dumps(return_json, separators=(",", ":")),
                 content_type="application/json",
@@ -167,4 +196,4 @@ class OTAHandler(BaseHandler):
             response = web.Response(text="OTA接口异常", content_type="text/plain")
         finally:
             self._add_cors_headers(response)
-            return response
\ No newline at end of file
+            return response
diff --git a/main/xiaozhi-server/core/auth.py b/main/xiaozhi-server/core/auth.py
index ef643132..be95da21 100755
--- a/main/xiaozhi-server/core/auth.py
+++ b/main/xiaozhi-server/core/auth.py
@@ -1,54 +1,72 @@
-from config.logger import setup_logging
-
-TAG = __name__
-logger = setup_logging()
+import hmac
+import base64
+import hashlib
+import time
 
 
 class AuthenticationError(Exception):
     """认证异常"""
+
     pass
 
 
-class AuthMiddleware:
-    def __init__(self, config):
-        self.config = config
-        self.auth_config = config["server"].get("auth", {})
-        # 构建token查找表
-        self.tokens = {
-            item["token"]: item["name"]
-            for item in self.auth_config.get("tokens", [])
-        }
-        # 设备白名单
-        self.allowed_devices = set(
-            self.auth_config.get("allowed_devices", [])
-        )
+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
 
-    async def authenticate(self, headers):
-        """验证连接请求"""
-        # 检查是否启用认证
-        if not self.auth_config.get("enabled", False):
             return True
-
-        # 检查设备是否在白名单中
-        device_id = headers.get("device-id", "")
-
-        if self.allowed_devices and device_id in self.allowed_devices:
-            return True
-
-        # 验证Authorization header
-        auth_header = headers.get("authorization", "")
-        if not auth_header.startswith("Bearer "):
-            logger.bind(tag=TAG).error("Missing or invalid Authorization header")
-            raise AuthenticationError("Missing or invalid Authorization header")
-
-        token = auth_header.split(" ")[1]
-        if token not in self.tokens:
-            logger.bind(tag=TAG).error(f"Invalid token: {token}")
-            raise AuthenticationError("Invalid token")
-
-        logger.bind(tag=TAG).info(f"Authentication successful - Device: {device_id}, Token: {self.tokens[token]}")
-        return True
-
-    def get_token_name(self, token):
-        """获取token对应的设备名称"""
-        return self.tokens.get(token)
+        except Exception:
+            return False
diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py
index 8db70fbf..4337a488 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,
@@ -31,8 +32,8 @@ from core.providers.asr.dto.dto import InterfaceType
 from core.handle.textHandle import handleTextMessage
 from core.providers.tools.unified_tool_handler import UnifiedToolHandler
 from plugins_func.loadplugins import auto_import_modules
-from plugins_func.register import Action, ActionResponse
-from core.auth import AuthMiddleware, AuthenticationError
+from plugins_func.register import Action
+from core.auth import AuthenticationError
 from config.config_loader import get_private_config_from_api
 from core.providers.tts.dto.dto import ContentType, TTSMessageDTO, SentenceType
 from config.logger import setup_logging, build_module_string, create_connection_logger
@@ -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)
@@ -164,25 +164,6 @@ class ConnectionHandler:
         try:
             # 获取并验证headers
             self.headers = dict(ws.request.headers)
-
-            if self.headers.get("device-id", None) is None:
-                # 尝试从 URL 的查询参数中获取 device-id
-                from urllib.parse import parse_qs, urlparse
-
-                # 从 WebSocket 请求中获取路径
-                request_path = ws.request.path
-                if not request_path:
-                    self.logger.bind(tag=TAG).error("无法获取请求路径")
-                    return
-                parsed_url = urlparse(request_path)
-                query_params = parse_qs(parsed_url.query)
-                if "device-id" in query_params:
-                    self.headers["device-id"] = query_params["device-id"][0]
-                    self.headers["client-id"] = query_params["client-id"][0]
-                else:
-                    await ws.send("端口正常，如需测试连接，请使用test_page.html")
-                    await self.close(ws)
-                    return
             real_ip = self.headers.get("x-real-ip") or self.headers.get(
                 "x-forwarded-for"
             )
@@ -194,12 +175,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/websocket_server.py b/main/xiaozhi-server/core/websocket_server.py
index 09ba7961..c0c88fc5 100644
--- a/main/xiaozhi-server/core/websocket_server.py
+++ b/main/xiaozhi-server/core/websocket_server.py
@@ -1,8 +1,11 @@
 import asyncio
+import json
+
 import websockets
 from config.logger import setup_logging
 from core.connection import ConnectionHandler
 from config.config_loader import get_config_from_api
+from core.auth import AuthManager, AuthenticationError
 from core.utils.modules_initialize import initialize_modules
 from core.utils.util import check_vad_update, check_asr_update
 
@@ -32,6 +35,14 @@ 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 = self.config["server"]["auth_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")
@@ -43,7 +54,40 @@ class WebSocketServer:
             await asyncio.Future()
 
     async def _handle_connection(self, websocket):
+        headers = dict(websocket.request.headers)
+        if headers.get("device-id", None) is None:
+            # 尝试从 URL 的查询参数中获取 device-id
+            from urllib.parse import parse_qs, urlparse
+
+            # 从 WebSocket 请求中获取路径
+            request_path = websocket.request.path
+            if not request_path:
+                self.logger.bind(tag=TAG).error("无法获取请求路径")
+                await websocket.close()
+                return
+            parsed_url = urlparse(request_path)
+            query_params = parse_qs(parsed_url.query)
+            if "device-id" not in query_params:
+                await websocket.send("端口正常，如需测试连接，请使用test_page.html")
+                await websocket.close()
+                return
+            else:
+                websocket.request.headers["device-id"] = query_params["device-id"][0]
+            if "client-id" in query_params:
+                websocket.request.headers["client-id"] = query_params["client-id"][0]
+            if "authorization" in query_params:
+                websocket.request.headers["authorization"] = query_params[
+                    "authorization"
+                ][0]
+
         """处理新连接，每次创建独立的ConnectionHandler"""
+        # 先认证，后建立连接
+        try:
+            await self._handle_auth(websocket)
+        except AuthenticationError:
+            await websocket.send("认证失败")
+            await websocket.close()
+            return
         # 创建ConnectionHandler时传入当前server实例
         handler = ConnectionHandler(
             self.config,
@@ -136,3 +180,26 @@ 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 in self.allowed_devices:
+                # 如果属于白名单内的设备，不校验token，直接放行
+                return
+            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")