's'

crypto_utils.py

@@ -43,24 +43,36 @@ def aes_decrypt_base64(cipher_b64: str, encoding_aes_key: str) -> str:
     cipher = AES.new(aes_key, AES.MODE_CBC, iv)
     decrypted = cipher.decrypt(cipher_bytes)
     decrypted = pkcs5_unpadding(decrypted)
-    # 微信消息体格式：16B随机串 + 4B网络序长度 + 明文 + appid
-    if len(decrypted) < 20:
-        raise ValueError("解密后消息长度非法")
-    content = decrypted[16:]
-    msg_len = struct.unpack("!I", content[:4])[0]
-    msg_start = 4
-    msg_end = msg_start + msg_len
-    if msg_end > len(content):
-        raise ValueError("解密后消息长度字段越界")
-    msg = content[msg_start:msg_end]
-    return msg.decode("utf-8")
+    # 兼容两种格式：
+    # 1) 微信消息体：16B随机串 + 4B长度 + 明文 + appid
+    # 2) 纯文本：直接是明文 JSON/XML
+    try:
+        if len(decrypted) < 20:
+            raise ValueError("payload too short")
+        content = decrypted[16:]
+        msg_len = struct.unpack("!I", content[:4])[0]
+        msg_start = 4
+        msg_end = msg_start + msg_len
+        if msg_end > len(content):
+            raise ValueError("msg length out of range")
+        msg = content[msg_start:msg_end]
+        text = msg.decode("utf-8")
+        if text:
+            return text
+    except Exception:
+        pass
+    return decrypted.decode("utf-8")
 
 
-def aes_encrypt_base64(plaintext: str, encoding_aes_key: str, app_id: str) -> str:
+def aes_encrypt_base64(plaintext: str, encoding_aes_key: str, app_id: str | None = None) -> str:
     aes_key, iv = get_aes_key_and_iv(encoding_aes_key)
     raw = plaintext.encode("utf-8")
-    # 微信消息体格式：16B随机串 + 4B网络序长度 + 明文 + appid
-    data = os.urandom(16) + struct.pack("!I", len(raw)) + raw + app_id.encode("utf-8")
+    if app_id:
+        # 微信消息体格式：16B随机串 + 4B网络序长度 + 明文 + appid
+        data = os.urandom(16) + struct.pack("!I", len(raw)) + raw + app_id.encode("utf-8")
+    else:
+        # 纯明文格式
+        data = raw
     padded = pkcs5_padding(data)
     cipher = AES.new(aes_key, AES.MODE_CBC, iv)
     encrypted = cipher.encrypt(padded)