#!/usr/bin/env python3 """ 企业微信会话存档解密脚本 使用RSA私钥解密加密的会话数据 """ import json import os import sys import base64 from pathlib import Path from datetime import datetime from dotenv import load_dotenv # 加载环境变量 load_dotenv() # 添加当前目录到路径，以便导入wework_ctypes sys.path.insert(0, str(Path(__file__).parent)) from wework_ctypes import WeWorkFinanceSDK, WeWorkSDKError def load_rsa_private_key(key_path: str) -> str: """ 加载RSA私钥 Args: key_path: 私钥文件路径 Returns: 私钥内容字符串 """ try: with open(key_path, 'r') as f: key_content = f.read() return key_content except Exception as e: raise Exception(f"加载RSA私钥失败: {e}") def rsa_decrypt_base64(encrypted_data_base64: str, private_key: str) -> str: """ 使用RSA私钥解密Base64编码的数据 Args: encrypted_data_base64: Base64编码的加密数据 private_key: RSA私钥字符串 Returns: 解密后的字符串 """ try: # 使用cryptography库进行RSA解密 from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import padding from cryptography.hazmat.backends import default_backend import base64 # 加载私钥 private_key_obj = serialization.load_pem_private_key( private_key.encode('utf-8'), password=None, backend=default_backend() ) # Base64解码 encrypted_data = base64.b64decode(encrypted_data_base64) # RSA解密 (使用PKCS1v15填充) decrypted_data = private_key_obj.decrypt( encrypted_data, padding.PKCS1v15() ) return decrypted_data.decode('utf-8') except ImportError: raise Exception("需要安装cryptography库: pip install cryptography") except Exception as e: raise Exception(f"RSA解密失败: {e}") def decrypt_chat_message(encrypt_random_key: str, encrypt_chat_msg: str, rsa_private_key_path: str, publickey_ver: int = None) -> tuple: """ 解密聊天消息 Args: encrypt_random_key: 加密的随机密钥（Base64编码） encrypt_chat_msg: 加密的聊天消息 rsa_private_key_path: RSA私钥文件路径或目录 publickey_ver: publickey_ver版本号，用于动态匹配私钥 Returns: (解密后的消息字典, 实际使用的私钥路径) """ print(f"正在解密消息...") print(f"加密密钥长度: {len(encrypt_random_key)}") print(f"加密消息长度: {len(encrypt_chat_msg)}") # 动态确定私钥路径 actual_key_path = rsa_private_key_path # 检查是否使用默认私钥路径（用户未指定自定义路径） # 通过检查路径是否指向默认文件来判断 default_key_path = Path("cert/1_rsa.key") specified_key_path = Path(rsa_private_key_path) # 如果使用默认私钥路径且提供了publickey_ver，则动态匹配 if publickey_ver is not None and specified_key_path == default_key_path: # 构建动态私钥路径：cert/{publickey_ver}_rsa.key dynamic_key_path = f"cert/{publickey_ver}_rsa.key" dynamic_path = Path(dynamic_key_path) if dynamic_path.exists(): actual_key_path = dynamic_key_path print(f"检测到publickey_ver={publickey_ver}，使用动态私钥: {actual_key_path}") else: print(f"警告: 动态私钥文件不存在: {dynamic_key_path}，使用默认私钥: {actual_key_path}") # 1. 加载RSA私钥 print(f"加载RSA私钥: {actual_key_path}") private_key = load_rsa_private_key(actual_key_path) print(f"✓ RSA私钥加载成功") # 2. 使用RSA私钥解密encrypt_random_key得到encrypt_key print("正在使用RSA解密随机密钥...") try: encrypt_key = rsa_decrypt_base64(encrypt_random_key, private_key) print(f"✓ RSA解密成功，encrypt_key长度: {len(encrypt_key)}") except Exception as e: print(f"✗ RSA解密失败: {e}") # 尝试直接使用encrypt_random_key作为encrypt_key（某些情况下可能已经解密） print("尝试使用原始encrypt_random_key作为encrypt_key...") encrypt_key = encrypt_random_key # 3. 使用SDK解密聊天消息 print("正在使用SDK解密聊天消息...") try: # 初始化SDK（只需要解密功能，不需要corpid和secret） with WeWorkFinanceSDK("dummy", "dummy") as sdk: decrypted_msg = sdk.decrypt_data(encrypt_key, encrypt_chat_msg) print("✓ SDK解密成功") # 尝试解析JSON try: result = json.loads(decrypted_msg) return result, actual_key_path except json.JSONDecodeError: # 如果不是JSON，返回原始文本 return {"raw_message": decrypted_msg}, actual_key_path except WeWorkSDKError as e: raise Exception(f"SDK解密失败: {e}") except Exception as e: raise Exception(f"解密过程出错: {e}") def decrypt_file(input_file: str, rsa_private_key_path: str, output_file: str = None): """ 解密文件中的聊天数据 Args: input_file: 输入文件路径（JSON格式） rsa_private_key_path: RSA私钥文件路径 output_file: 输出文件路径（可选） """ print("=" * 60) print("企业微信会话存档解密脚本") print("=" * 60) # 检查输入文件 input_path = Path(input_file) if not input_path.exists(): print(f"错误: 输入文件不存在: {input_file}") return 1 # 检查RSA私钥文件（只检查用户指定的路径，动态路径在解密时检查） rsa_path = Path(rsa_private_key_path) if not rsa_path.exists(): print(f"错误: 指定的RSA私钥文件不存在: {rsa_private_key_path}") return 1 print(f"输入文件: {input_path.absolute()}") print(f"指定私钥: {rsa_path.absolute()}") print("-" * 60) # 用于记录实际使用的私钥路径 actual_key_paths_used = set() try: # 读取输入文件 with open(input_path, 'r', encoding='utf-8') as f: data = json.load(f) print(f"文件加载成功") # 检查数据格式 if "chatdata" not in data: print("错误: 文件中没有chatdata字段") return 1 chatdata = data.get("chatdata", []) chat_count = len(chatdata) print(f"共发现 {chat_count} 条聊天记录") if chat_count == 0: print("没有可解密的聊天记录") return 0 # 解密所有消息 decrypted_messages = [] success_count = 0 fail_count = 0 for i, msg in enumerate(chatdata): # print('msg', type(msg)) seq = msg.get("seq", i + 1) msgid = msg.get("msgid", f"msg_{i}") encrypt_random_key = msg.get("encrypt_random_key", "") encrypt_chat_msg = msg.get("encrypt_chat_msg", "") print(f"\n[{i+1}/{chat_count}] 解密消息 seq={seq}, msgid={msgid[:20]}...") if not encrypt_random_key or not encrypt_chat_msg: print(f" 跳过: 缺少加密数据") fail_count += 1 decrypted_messages.append({ "seq": seq, "msgid": msgid, "error": "缺少加密数据", "original": msg }) continue try: # 获取publickey_ver publickey_ver = msg.get("publickey_ver") # 解密消息 decrypted_msg, actual_key_path = decrypt_chat_message( encrypt_random_key, encrypt_chat_msg, rsa_private_key_path, publickey_ver ) # 记录实际使用的私钥路径 actual_key_paths_used.add(actual_key_path) # 添加元数据 if isinstance(decrypted_msg, dict): decrypted_msg["_metadata"] = { "seq": seq, "msgid": msgid, "decrypt_time": datetime.now().isoformat(), "original_has_encryption": True } decrypted_messages.append({ "seq": seq, "msgid": msgid, "decrypted": decrypted_msg, "original_summary": { "publickey_ver": msg.get("publickey_ver"), "encrypt_key_len": len(encrypt_random_key), "encrypt_msg_len": len(encrypt_chat_msg) } }) success_count += 1 print(f" ✓ 解密成功") # 显示解密后的消息摘要 if isinstance(decrypted_msg, dict): msg_type = decrypted_msg.get("msgtype", "unknown") from_user = decrypted_msg.get("from", "unknown") to_user = decrypted_msg.get("tolist", ['unknown'])[0] if isinstance(decrypted_msg.get("tolist"), list) and len(decrypted_msg.get("tolist")) > 1: to_user += f" 等 {len(decrypted_msg.get('tolist'))} 人" print(f" 类型: {msg_type}, 来自: {from_user}, 发送给: {to_user}") # 成功解密后继续处理下一条消息 continue except Exception as e: print(f" ✗ 解密失败: {e}") fail_count += 1 decrypted_messages.append({ "seq": seq, "msgid": msgid, "error": str(e), "original": msg }) print("\n" + "-" * 60) print(f"解密完成: 成功 {success_count} 条, 失败 {fail_count} 条") # 显示实际使用的私钥信息 if actual_key_paths_used: print(f"实际使用的私钥文件:") for key_path in sorted(actual_key_paths_used): print(f" - {key_path}") # 生成输出文件名 if output_file is None: timestamp = datetime.now().strftime("%y%m%d_%H%M%S") output_file = f"data/decrypted_{input_path.stem}_{timestamp}.json" output_path = Path(output_file) # 准备输出数据 output_data = { "metadata": { "input_file": str(input_path.absolute()), "decrypt_time": datetime.now().isoformat(), "total_messages": chat_count, "successful_decrypts": success_count, "failed_decrypts": fail_count, "specified_key": str(rsa_path.absolute()), "actual_keys_used": sorted([str(Path(p).absolute()) for p in actual_key_paths_used]) }, "decrypted_messages": decrypted_messages, "original_metadata": data.get("_metadata", {}) } # 保存解密结果 with open(output_path, 'w', encoding='utf-8') as f: json.dump(output_data, f, indent=2, ensure_ascii=False) print(f"解密结果已保存到: {output_path.absolute()}") # 显示一些解密后的消息示例 if success_count > 0: print("\n解密消息示例:") for i, msg in enumerate(decrypted_messages[:3]): # 显示前3条 if "decrypted" in msg: decrypted = msg["decrypted"] if isinstance(decrypted, dict): seq = msg.get("seq", "N/A") msg_type = decrypted.get("msgtype", "unknown") content = decrypted.get("content", "无内容")[:100] print(f" [{i+1}] seq={seq}, 类型={msg_type}") print(f" 内容: {content}...") print("\n" + "=" * 60) print("解密完成") return 0 except json.JSONDecodeError as e: print(f"JSON解析错误: {e}") return 1 except Exception as e: print(f"解密过程出错: {e}") import traceback traceback.print_exc() return 1 def main(): """主函数""" import argparse parser = argparse.ArgumentParser( description="企业微信会话存档解密脚本", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 使用示例: # 解密data目录下的文件 %(prog)s data/chat_0_100_251203.json # 指定RSA私钥路径 %(prog)s data/chat_0_100_251203.json --key /path/to/private_key.pem # 指定输出文件 %(prog)s data/chat_0_100_251203.json --output decrypted_result.json """ ) parser.add_argument('input_file', help='输入文件路径（JSON格式）') parser.add_argument('--key', '-k', default='cert/1_rsa.key', help='RSA私钥文件路径，默认: cert/1_rsa.key') parser.add_argument('--output', '-o', help='输出文件路径，默认自动生成') args = parser.parse_args() return decrypt_file(args.input_file, args.key, args.output) if __name__ == "__main__": sys.exit(main())