Merge pull request #1570 from xinnan-tech/hot-fix

update:优化视觉接口认证校验

main/xiaozhi-server/core/utils/auth.py

@@ -1,12 +1,80 @@
 import jwt
 import time
+import json
+import os
 from datetime import datetime, timedelta, timezone
-from typing import Optional, Tuple
+from typing import Tuple, Optional
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+from cryptography.hazmat.primitives import padding
+from cryptography.hazmat.backends import default_backend
+import base64
 
 
 class AuthToken:
     def __init__(self, secret_key: str):
-        self.secret_key = secret_key
+        self.secret_key = secret_key.encode()  # 转换为字节
+        # 从密钥派生固定长度的加密密钥 (32字节 for AES-256)
+        self.encryption_key = self._derive_key(32)
+
+    def _derive_key(self, length: int) -> bytes:
+        """派生固定长度的密钥"""
+        from cryptography.hazmat.primitives import hashes
+        from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
+
+        # 使用固定盐值（实际生产环境应使用随机盐）
+        salt = b"fixed_salt_placeholder"  # 生产环境应改为随机生成
+        kdf = PBKDF2HMAC(
+            algorithm=hashes.SHA256(),
+            length=length,
+            salt=salt,
+            iterations=100000,
+            backend=default_backend(),
+        )
+        return kdf.derive(self.secret_key)
+
+    def _encrypt_payload(self, payload: dict) -> str:
+        """使用AES-GCM加密整个payload"""
+        # 将payload转换为JSON字符串
+        payload_json = json.dumps(payload)
+
+        # 生成随机IV
+        iv = os.urandom(12)
+        # 创建加密器
+        cipher = Cipher(
+            algorithms.AES(self.encryption_key),
+            modes.GCM(iv),
+            backend=default_backend(),
+        )
+        encryptor = cipher.encryptor()
+
+        # 加密并生成标签
+        ciphertext = encryptor.update(payload_json.encode()) + encryptor.finalize()
+        tag = encryptor.tag
+
+        # 组合 IV + 密文 + 标签
+        encrypted_data = iv + ciphertext + tag
+        return base64.urlsafe_b64encode(encrypted_data).decode()
+
+    def _decrypt_payload(self, encrypted_data: str) -> dict:
+        """解密AES-GCM加密的payload"""
+        # 解码Base64
+        data = base64.urlsafe_b64decode(encrypted_data.encode())
+        # 拆分组件
+        iv = data[:12]
+        tag = data[-16:]
+        ciphertext = data[12:-16]
+
+        # 创建解密器
+        cipher = Cipher(
+            algorithms.AES(self.encryption_key),
+            modes.GCM(iv, tag),
+            backend=default_backend(),
+        )
+        decryptor = cipher.decryptor()
+
+        # 解密
+        plaintext = decryptor.update(ciphertext) + decryptor.finalize()
+        return json.loads(plaintext.decode())
 
     def generate_token(self, device_id: str) -> str:
         """
@@ -17,11 +85,17 @@ class AuthToken:
         # 设置过期时间为1小时后
         expire_time = datetime.now(timezone.utc) + timedelta(hours=1)
 
-        # 创建payload
+        # 创建原始payload
         payload = {"device_id": device_id, "exp": expire_time.timestamp()}
 
+        # 加密整个payload
+        encrypted_payload = self._encrypt_payload(payload)
+
+        # 创建外层payload，包含加密数据
+        outer_payload = {"data": encrypted_payload}
+
         # 使用JWT进行编码
-        token = jwt.encode(payload, self.secret_key, algorithm="HS256")
+        token = jwt.encode(outer_payload, self.secret_key, algorithm="HS256")
         return token
 
     def verify_token(self, token: str) -> Tuple[bool, Optional[str]]:
@@ -31,13 +105,22 @@ class AuthToken:
         :return: (是否有效, 设备ID)
         """
         try:
-            # 解码token
-            payload = jwt.decode(token, self.secret_key, algorithms=["HS256"])
+            # 先验证外层JWT（签名和过期时间）
+            outer_payload = jwt.decode(token, self.secret_key, algorithms=["HS256"])
 
-            # 检查是否过期
-            if payload["exp"] < time.time():
+            # 解密内层payload
+            inner_payload = self._decrypt_payload(outer_payload["data"])
+
+            # 再次检查过期时间（双重验证）
+            if inner_payload["exp"] < time.time():
                 return False, None
 
-            return True, payload["device_id"]
+            return True, inner_payload["device_id"]
+
         except jwt.InvalidTokenError:
             return False, None
+        except json.JSONDecodeError:
+            return False, None
+        except Exception as e:  # 捕获其他可能的错误
+            print(f"Token verification failed: {str(e)}")
+            return False, None