From 4fd294dc80cff649969badcd7d7e10c1276d1991 Mon Sep 17 00:00:00 2001 From: hrz <1710360675@qq.com> Date: Mon, 13 Oct 2025 17:52:28 +0800 Subject: [PATCH] =?UTF-8?q?update=EF=BC=9A=201=E3=80=81=E6=8A=8Aota=5Fauth?= =?UTF-8?q?=E5=86=85=E5=AE=B9=E5=90=88=E5=B9=B6=E5=88=B0auth.py=202?= =?UTF-8?q?=E3=80=81=E7=B2=BE=E7=AE=80auth=E5=AF=86=E9=92=A5=EF=BC=8C?= =?UTF-8?q?=E7=AE=80=E5=8C=96=E9=83=A8=E7=BD=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/config.yaml | 16 +-- main/xiaozhi-server/core/api/ota_handler.py | 52 ++++----- main/xiaozhi-server/core/auth.py | 108 +++++++++++-------- main/xiaozhi-server/core/connection.py | 23 +--- main/xiaozhi-server/core/ota_auth.py | 64 ----------- main/xiaozhi-server/core/websocket_server.py | 54 +++++++--- 6 files changed, 133 insertions(+), 184 deletions(-) delete mode 100644 main/xiaozhi-server/core/ota_auth.py diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml index 261f68cf..b2cc9fe6 100644 --- a/main/xiaozhi-server/config.yaml +++ b/main/xiaozhi-server/config.yaml @@ -31,18 +31,10 @@ server: auth: # 是否启用认证 enabled: false - # token签名秘钥 - signature_key: "your-token-signature-key" - # 设备的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 44eee43b..d3d37348 100644 --- a/main/xiaozhi-server/core/api/ota_handler.py +++ b/main/xiaozhi-server/core/api/ota_handler.py @@ -5,7 +5,7 @@ import hashlib import hmac from aiohttp import web -from core.ota_auth import AuthManager +from core.auth import AuthManager from core.utils.util import get_local_ip from core.api.base_handler import BaseHandler @@ -18,27 +18,27 @@ class OTAHandler(BaseHandler): 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", "") + 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 "" @@ -97,10 +97,9 @@ class OTAHandler(BaseHandler): "url": "", }, } - mqtt_gateway_endpoint = server_config.get("mqtt_gateway") - + if mqtt_gateway_endpoint: # 如果配置了非空字符串 # 尝试从请求数据中获取设备型号 device_model = "default" @@ -118,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 = "" @@ -132,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: @@ -145,27 +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 in 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 + "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", @@ -194,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 bd14a6fd..4337a488 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -32,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 @@ -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" ) diff --git a/main/xiaozhi-server/core/ota_auth.py b/main/xiaozhi-server/core/ota_auth.py deleted file mode 100644 index 3c5e7d9d..00000000 --- a/main/xiaozhi-server/core/ota_auth.py +++ /dev/null @@ -1,64 +0,0 @@ -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 b3a5a7d9..c0c88fc5 100644 --- a/main/xiaozhi-server/core/websocket_server.py +++ b/main/xiaozhi-server/core/websocket_server.py @@ -3,10 +3,9 @@ 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.auth import AuthManager, AuthenticationError from core.utils.modules_initialize import initialize_modules from core.utils.util import check_vad_update, check_asr_update @@ -39,10 +38,8 @@ class WebSocketServer: 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", "") + 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) @@ -57,17 +54,38 @@ 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(json.dumps({ - "type": "alert", - "status": "ERROR", - "message": "认证失败", - "emotion": "sad" - })) + await websocket.send("认证失败") await websocket.close() return # 创建ConnectionHandler时传入当前server实例 @@ -169,17 +187,19 @@ class WebSocketServer: 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") + if self.allowed_devices and device_id in self.allowed_devices: + # 如果属于白名单内的设备，不校验token，直接放行 + return else: # 否则校验token token = headers.get("authorization", "") - if token.startswith('Bearer '): + 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) + auth_success = self.auth.verify_token( + token, client_id=client_id, username=device_id + ) if not auth_success: raise AuthenticationError("Invalid token")