87 lines
4.1 KiB
Python
87 lines
4.1 KiB
Python
"""Bounded, read-only scan of the local WeChat process. No system changes."""
|
|
import ctypes as C
|
|
import json
|
|
import platform
|
|
import re
|
|
import subprocess
|
|
import time
|
|
|
|
|
|
def wechat_pid():
|
|
if platform.system() != 'Darwin':
|
|
return None
|
|
result = subprocess.run(['/usr/bin/pgrep', '-x', 'WeChat'], capture_output=True, text=True, timeout=5)
|
|
for value in result.stdout.split():
|
|
pid = int(value)
|
|
lib = C.CDLL('/usr/lib/libproc.dylib')
|
|
lib.proc_pidpath.argtypes = [C.c_int, C.c_void_p, C.c_uint32]
|
|
buf = C.create_string_buffer(4096)
|
|
if lib.proc_pidpath(pid, buf, len(buf)) > 0:
|
|
path = buf.value.decode('utf-8', errors='replace')
|
|
if path.endswith('/WeChat.app/Contents/MacOS/WeChat'):
|
|
return pid
|
|
return None
|
|
|
|
|
|
def scan_keys():
|
|
if platform.system() != 'Darwin':
|
|
raise ValueError('请在安装微信的 Mac 上运行本地服务')
|
|
pid = wechat_pid()
|
|
if not pid:
|
|
raise ValueError('未找到运行中的微信，请先打开并登录微信')
|
|
lib = C.CDLL('/usr/lib/libSystem.B.dylib')
|
|
own = C.c_uint.in_dll(lib, "mach_task_self_").value
|
|
task = C.c_uint()
|
|
lib.task_for_pid.argtypes = [C.c_uint, C.c_int, C.POINTER(C.c_uint)]
|
|
if lib.task_for_pid(own, pid, C.byref(task)):
|
|
raise ValueError('系统拒绝读取微信进程。管理员授权也可能受微信签名限制；工具不会修改签名或关闭 SIP。可导入本地密钥文件继续。')
|
|
class Region(C.Structure):
|
|
_pack_ = 4
|
|
_fields_ = [('protection', C.c_int), ('max_protection', C.c_int), ('inheritance', C.c_int),
|
|
('shared', C.c_uint), ('reserved', C.c_uint), ('offset', C.c_uint64),
|
|
('behavior', C.c_int), ('wired', C.c_ushort)]
|
|
lib.mach_vm_region.argtypes = [C.c_uint, C.POINTER(C.c_uint64), C.POINTER(C.c_uint64), C.c_int,
|
|
C.c_void_p, C.POINTER(C.c_uint), C.POINTER(C.c_uint)]
|
|
lib.mach_vm_read_overwrite.argtypes = [C.c_uint, C.c_uint64, C.c_uint64, C.c_void_p, C.POINTER(C.c_uint64)]
|
|
lib.mach_port_deallocate.argtypes = [C.c_uint, C.c_uint]
|
|
address, read, keys = 0, 0, set()
|
|
deadline = time.monotonic() + 45
|
|
pattern = re.compile(rb"[xX]'([0-9a-fA-F]{96}|[0-9a-fA-F]{64})'")
|
|
try:
|
|
while read < 2 * 1024**3 and time.monotonic() < deadline and len(keys) < 2000:
|
|
addr, size, region, count, obj = C.c_uint64(address), C.c_uint64(), Region(), C.c_uint(C.sizeof(Region)//4), C.c_uint()
|
|
code = lib.mach_vm_region(task, C.byref(addr), C.byref(size), 9, C.byref(region), C.byref(count), C.byref(obj))
|
|
if obj.value:
|
|
lib.mach_port_deallocate(own, obj)
|
|
if code or not size.value:
|
|
break
|
|
end = addr.value + size.value
|
|
if end <= address:
|
|
break
|
|
if region.protection & 1 and region.protection & 2 and not region.protection & 4:
|
|
pos, tail = addr.value, b''
|
|
while pos < end and read < 2 * 1024**3 and time.monotonic() < deadline:
|
|
length = min(1024**2, end-pos)
|
|
buf, actual = C.create_string_buffer(length), C.c_uint64()
|
|
if not lib.mach_vm_read_overwrite(task, pos, length, buf, C.byref(actual)):
|
|
chunk = tail + buf.raw[:actual.value]
|
|
keys.update(m[1].decode().lower() for m in pattern.finditer(chunk))
|
|
tail = chunk[-100:]
|
|
else:
|
|
tail = b''
|
|
read += length
|
|
pos += length
|
|
address = end
|
|
finally:
|
|
lib.mach_port_deallocate(own, task)
|
|
if not keys:
|
|
raise ValueError('扫描完成但未发现可用格式的密钥。请在微信中打开目标聊天后重试，或导入已有密钥文件。该版本可能需要进一步适配。')
|
|
return sorted(keys)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
print(json.dumps({'keys': scan_keys()}))
|
|
except Exception as exc:
|
|
print(json.dumps({'error': str(exc) if isinstance(exc, ValueError) else '本机密钥扫描失败'}))
|