136 lines
6.4 KiB
Python
136 lines
6.4 KiB
Python
"""Bounded, read-only scan of the local WeChat process. No system changes."""
|
|
import ctypes as C
|
|
import json
|
|
import os
|
|
import platform
|
|
import re
|
|
import subprocess
|
|
import time
|
|
|
|
|
|
def wechat_processes():
|
|
if platform.system() != 'Darwin':
|
|
return []
|
|
result = subprocess.run(['/usr/bin/pgrep', '-x', 'WeChat'], capture_output=True, text=True, timeout=5)
|
|
processes = []
|
|
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'):
|
|
processes.append({'pid': pid, 'path': path})
|
|
return processes
|
|
|
|
|
|
def wechat_pid():
|
|
processes = wechat_processes()
|
|
return processes[0]['pid'] if processes else None
|
|
|
|
|
|
def signing_flags(pid):
|
|
lib = C.CDLL('/usr/lib/libSystem.B.dylib', use_errno=True)
|
|
lib.csops.argtypes = [C.c_int, C.c_uint, C.c_void_p, C.c_size_t]
|
|
value = C.c_uint32()
|
|
if lib.csops(pid, 0, C.byref(value), C.sizeof(value)):
|
|
return {'hardened_runtime': None, 'get_task_allow': None}
|
|
return {'hardened_runtime': bool(value.value & 0x10000), 'get_task_allow': bool(value.value & 4)}
|
|
|
|
|
|
def permission_diagnostic(code=None, pid=None):
|
|
processes = wechat_processes()
|
|
target = next((p for p in processes if p['pid'] == pid), processes[0] if len(processes) == 1 else None)
|
|
result = {'platform': platform.system(), 'architecture': platform.machine(),
|
|
'administrator': hasattr(os, 'geteuid') and os.geteuid() == 0,
|
|
'task_for_pid_code': code, 'processes': processes, 'code': 'no_process',
|
|
'summary': '请先打开并登录微信，再检查读取权限。'}
|
|
if len(processes) > 1:
|
|
result.update(code='multiple_processes', summary='检测到多个微信进程，请仅保留要读取的一个微信。')
|
|
elif target:
|
|
result.update(signing_flags(target['pid']))
|
|
if result['hardened_runtime'] and result['get_task_allow'] is False:
|
|
result.update(code='target_restricted', summary='微信启用了强化运行时，且没有允许调试。普通和管理员读取都可能被签名策略拒绝；反复输入密码不能改变此限制。')
|
|
elif code is not None:
|
|
result.update(code='access_denied', summary='macOS 拒绝读取进程，请检查系统“隐私与安全性 → 开发者工具”中的终端授权。权限设置不保证能突破目标签名限制。')
|
|
else:
|
|
result.update(code='can_attempt', summary='未检测到明确的目标签名阻塞，可以尝试读取；实际结果以系统返回为准。')
|
|
return result
|
|
|
|
|
|
class AccessDenied(ValueError):
|
|
def __init__(self, diagnostic):
|
|
self.diagnostic = diagnostic
|
|
super().__init__(diagnostic['summary'])
|
|
|
|
|
|
def scan_keys():
|
|
if platform.system() != 'Darwin':
|
|
raise ValueError('请在安装微信的 Mac 上运行本地服务')
|
|
processes = wechat_processes()
|
|
if len(processes) > 1:
|
|
raise AccessDenied(permission_diagnostic())
|
|
pid = processes[0]['pid'] if processes else None
|
|
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)]
|
|
code = lib.task_for_pid(own, pid, C.byref(task))
|
|
if code:
|
|
raise AccessDenied(permission_diagnostic(code, pid))
|
|
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:
|
|
payload = {'error': str(exc) if isinstance(exc, ValueError) else '本机密钥扫描失败'}
|
|
if isinstance(exc, AccessDenied):
|
|
payload['diagnostic'] = exc.diagnostic
|
|
print(json.dumps(payload))
|