3232
3333# ====================== 以下是hook逻辑 ======================================
3434
35- @dataclass
36- class HookConfig :
37- min_version : str
38- pattern : str
39- mask : str
40- offset : int
41- md5_pattern : str = ""
42- md5_mask : str = ""
43- md5_offset : int = 0
44-
4535class WeChatKeyFetcher :
4636 def __init__ (self ):
4737 self .process_name = "Weixin.exe"
4838 self .timeout_seconds = 60
4939
50- @staticmethod
51- def _hex_array_to_str (hex_array : List [int ]) -> str :
52- return " " .join ([f"{ b :02X} for b in hex_array ])
53-
54- def _get_hook_config (self , version_str : str ) -> Optional [HookConfig ]:
55- try :
56- v_curr = pkg_version .parse (version_str )
57- except Exception as e :
58- logger .error (f"版本号解析失败: { version_str } { e } )
59- return None
60-
61-
62- if v_curr > pkg_version .parse ("4.1.6.14" ):
63- return HookConfig (
64- min_version = ">4.1.6.14" ,
65- pattern = self ._hex_array_to_str ([
66- 0x24 , 0x50 , 0x48 , 0xC7 , 0x45 , 0x00 , 0xFE , 0xFF , 0xFF , 0xFF ,
67- 0x44 , 0x89 , 0xCF , 0x44 , 0x89 , 0xC3 , 0x49 , 0x89 , 0xD6 , 0x48 ,
68- 0x89 , 0xCE , 0x48 , 0x89
69- ]),
70- mask = "xxxxxxxxxxxxxxxxxxxxxxxx" ,
71- offset = - 3 ,
72- md5_pattern = "48 8D 4D 00 48 89 4D B0 48 89 45 B8 48 8D 7D 00 48 8D 55 B0 48 89 F9" ,
73- md5_mask = "xxx?xxxxxxxxxxx?xxxxxxx" ,
74- md5_offset = 4
75- )
76-
77- if pkg_version .parse ("4.1.4" ) <= v_curr <= pkg_version .parse ("4.1.6.14" ):
78- return HookConfig (
79- min_version = "4.1.4-4.1.6.14" ,
80- pattern = self ._hex_array_to_str ([
81- 0x24 , 0x08 , 0x48 , 0x89 , 0x6c , 0x24 , 0x10 , 0x48 , 0x89 , 0x74 ,
82- 0x00 , 0x18 , 0x48 , 0x89 , 0x7c , 0x00 , 0x20 , 0x41 , 0x56 , 0x48 ,
83- 0x83 , 0xec , 0x50 , 0x41
84- ]),
85- mask = "xxxxxxxxxx?xxxx?xxxxxxxx" ,
86- offset = - 3 ,
87- md5_pattern = "48 8D 4D 00 48 89 4D B0 48 89 45 B8 48 8D 7D 00 48 8D 55 B0 48 89 F9" ,
88- md5_mask = "xxx?xxxxxxxxxxx?xxxxxxx" ,
89- md5_offset = 4
90- )
91-
92- if v_curr < pkg_version .parse ("4.1.4" ):
93- """图片密钥可能是错的，版本过低没有测试"""
94- return HookConfig (
95- min_version = "<4.1.4" ,
96- pattern = self ._hex_array_to_str ([
97- 0x24 , 0x50 , 0x48 , 0xc7 , 0x45 , 0x00 , 0xfe , 0xff , 0xff , 0xff ,
98- 0x44 , 0x89 , 0xcf , 0x44 , 0x89 , 0xc3 , 0x49 , 0x89 , 0xd6 , 0x48 ,
99- 0x89 , 0xce , 0x48 , 0x89
100- ]),
101- mask = "xxxxxxxxxxxxxxxxxxxxxxxx" ,
102- offset = - 15 , # -0xf
103- md5_pattern = "48 8D 4D 00 48 89 4D B0 48 89 45 B8 48 8D 7D 00 48 8D 55 B0 48 89 F9" ,
104- md5_mask = "xxx?xxxxxxxxxxx?xxxxxxx" ,
105- md5_offset = 4
106- )
107-
108- return None
109-
11040 def kill_wechat (self ):
11141 """检测并查杀微信进程"""
11242 killed = False
@@ -125,17 +55,14 @@ def kill_wechat(self):
12555 def launch_wechat (self , exe_path : str ) -> int :
12656 """启动微信并返回 PID"""
12757 try :
128-
12958 process = subprocess .Popen (exe_path )
130-
13159 time .sleep (2 )
13260 candidates = []
13361 for proc in psutil .process_iter (['pid' , 'name' , 'create_time' ]):
13462 if proc .info ['name' ] == self .process_name :
13563 candidates .append (proc )
13664
13765 if candidates :
138-
13966 candidates .sort (key = lambda x : x .info ['create_time' ], reverse = True )
14067 target_pid = candidates [0 ].info ['pid' ]
14168 return target_pid
@@ -146,8 +73,8 @@ def launch_wechat(self, exe_path: str) -> int:
14673 logger .error (f"启动微信失败: { e } )
14774 raise RuntimeError (f"无法启动微信: { e } )
14875
149- def fetch_key (self ) -> dict :
150- """调用 wx_key 获取双密钥 """
76+ def fetch_db_key (self ) -> dict :
77+ """调用 wx_key 仅获取数据库密钥 (Hook 模式) """
15178 if wx_key is None :
15279 raise RuntimeError ("wx_key 模块未安装或加载失败" )
15380
@@ -160,36 +87,26 @@ def fetch_key(self) -> dict:
16087
16188 logger .info (f"Detect WeChat: { version } { exe_path } )
16289
163- config = self ._get_hook_config (version )
164- if not config :
165- raise RuntimeError (f"原生获取失败：当前微信版本 ({ version } )
166-
16790 self .kill_wechat ()
16891 pid = self .launch_wechat (exe_path )
16992 logger .info (f"WeChat launched, PID: { pid } )
17093
171- if not wx_key . initialize_hook ( pid , "" , config . pattern , config . mask , config . offset ,
172- config . md5_pattern , config . md5_mask , config . md5_offset ):
94+ # 仅传入 PID，触发数据库密钥自动 Hook
95+ if not wx_key . initialize_hook ( pid ):
17396 err = wx_key .get_last_error_msg ()
174- raise RuntimeError (f"Hook初始化失败 : { err } )
97+ raise RuntimeError (f"数据库 Hook 初始化失败 : { err } )
17598
17699 start_time = time .time ()
177100 found_db_key = None
178- found_md5_data = None
179101
180102 try :
181103 while True :
182104 if time .time () - start_time > self .timeout_seconds :
183- raise TimeoutError ("获取密钥超时 (60s)，请确保在弹出的微信中完成登录。" )
105+ raise TimeoutError ("获取数据库密钥超时 (60s)，请确保在弹出的微信中完成登录。" )
184106
185107 key_data = wx_key .poll_key_data ()
186- if key_data :
187- if 'key' in key_data :
188- found_db_key = key_data ['key' ]
189- if 'md5' in key_data :
190- found_md5_data = key_data ['md5' ]
191-
192- if found_db_key and found_md5_data :
108+ if key_data and 'key' in key_data :
109+ found_db_key = key_data ['key' ]
193110 break
194111
195112 while True :
@@ -204,22 +121,13 @@ def fetch_key(self) -> dict:
204121 logger .info ("Cleaning up hook..." )
205122 wx_key .cleanup_hook ()
206123
207- aes_key = None # gemini !!! ???
208- xor_key = None
209-
210- if found_md5_data and "|" in found_md5_data :
211- aes_key , xor_key_dec = found_md5_data .split ("|" )
212- xor_key = f"0x{ int (xor_key_dec ):02X}
213-
214124 return {
215- "db_key" : found_db_key ,
216- "aes_key" : aes_key ,
217- "xor_key" : xor_key
125+ "db_key" : found_db_key
218126 }
219127
220128def get_db_key_workflow ():
221129 fetcher = WeChatKeyFetcher ()
222- return fetcher .fetch_key ()
130+ return fetcher .fetch_db_key ()
223131
224132
225133# ============================== 以下是图片密钥逻辑 =====================================
@@ -232,6 +140,82 @@ def get_wechat_internal_global_config(wx_dir: Path, file_name1) -> bytes:
232140 return Path (target_path ).read_bytes ()
233141
234142
143+ def try_get_local_image_keys () -> List [Dict [str , Any ]]:
144+ """尝试通过本地算法提取图片密钥 (无需 Hook)"""
145+ if wx_key is None or not hasattr (wx_key , 'get_image_key' ):
146+ return []
147+
148+ try :
149+ res_json = wx_key .get_image_key ()
150+ if not res_json :
151+ return []
152+
153+ data = json .loads (res_json )
154+ accounts = data .get ('accounts' , [])
155+ results = []
156+ for acc in accounts :
157+ wxid = acc .get ('wxid' )
158+ keys = acc .get ('keys' , [])
159+ for k in keys :
160+ xor_key = k .get ('xorKey' )
161+ aes_key = k .get ('aesKey' )
162+ if xor_key is not None :
163+ results .append ({
164+ "wxid" : wxid ,
165+ "xor_key" : f"0x{ int (xor_key ):02X} ,
166+ "aes_key" : aes_key
167+ })
168+ return results
169+ except Exception as e :
170+ logger .error (f"本地提取图片密钥失败: { e } )
171+ return []
172+
173+
174+ async def get_image_key_integrated_workflow (account : Optional [str ] = None ) -> Dict [str , Any ]:
175+ """
176+ 集成图片密钥获取流程：
177+ 1. 优先尝试本地算法提取
178+ 2. 如果本地提取失败或未匹配到指定账号，尝试远程 API 解析
179+ """
180+ # 1. 尝试本地提取
181+ local_keys = try_get_local_image_keys ()
182+
183+ target_account_wxid = None
184+ if account :
185+ try :
186+ account_dir = _resolve_account_dir (account )
187+ wx_id_dir = _resolve_account_wxid_dir (account_dir )
188+ target_account_wxid = wx_id_dir .name
189+ except :
190+ target_account_wxid = account
191+
192+ if local_keys :
193+ # 如果指定了账号，尝试在本地结果中找匹配的
194+ if target_account_wxid :
195+ for k in local_keys :
196+ if k ['wxid' ] == target_account_wxid :
197+ logger .info (f"成功通过本地算法匹配到账号 { target_account_wxid } )
198+ upsert_account_keys_in_store (
199+ account = k ['wxid' ],
200+ image_xor_key = k ['xor_key' ],
201+ image_aes_key = k ['aes_key' ]
202+ )
203+ return k
204+ else :
205+ # 如果没指定账号，返回第一个发现的并存入 store (如果有的话)
206+ k = local_keys [0 ]
207+ logger .info (f"本地算法提取成功 (未指定账号，返回首个): { k ['wxid' ]} )
208+ upsert_account_keys_in_store (
209+ account = k ['wxid' ],
210+ image_xor_key = k ['xor_key' ],
211+ image_aes_key = k ['aes_key' ]
212+ )
213+ return k
214+
215+ # 2. 本地提取失败或不匹配，尝试远程解析
216+ logger .info ("本地算法提取未命中，尝试远程 API 解析..." )
217+ return await fetch_and_save_remote_keys (account )
218+
235219
236220async def fetch_and_save_remote_keys (account : Optional [str ] = None ) -> Dict [str , Any ]:
237221 account_dir = _resolve_account_dir (account )
0 commit comments