175 lines
10 KiB
Python
175 lines
10 KiB
Python
import json
|
|
import plistlib
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from wechat_tool.mac_connect import MacConnection
|
|
from wechat_tool.mac_keys import permission_diagnostic
|
|
from wechat_tool.mac_recovery import DebugRecovery
|
|
from wechat_tool.mac_guidance import next_step
|
|
|
|
|
|
class PermissionTests(unittest.TestCase):
|
|
def test_signed_target_is_distinct_from_administrator_permission(self):
|
|
proc = {'pid':123,'path':'/Applications/WeChat.app/Contents/MacOS/WeChat'}
|
|
with patch('wechat_tool.mac_keys.wechat_processes', return_value=[proc]), patch('wechat_tool.mac_keys.signing_flags', return_value={'hardened_runtime':True,'get_task_allow':False}):
|
|
result=permission_diagnostic(5,123)
|
|
self.assertEqual(result['code'],'target_restricted')
|
|
self.assertEqual(result['task_for_pid_code'],5)
|
|
self.assertTrue(result['attempted'])
|
|
with patch('wechat_tool.mac_keys.wechat_processes',return_value=[]):
|
|
self.assertEqual(permission_diagnostic()['code'],'no_process')
|
|
with patch('wechat_tool.mac_keys.wechat_processes',return_value=[proc,dict(proc,pid=456)]):
|
|
self.assertEqual(permission_diagnostic()['code'],'multiple_processes')
|
|
|
|
def test_signature_check_is_not_an_actual_read_failure(self):
|
|
proc={'pid':123,'path':'/Applications/WeChat.app/Contents/MacOS/WeChat'}
|
|
with patch('wechat_tool.mac_keys.wechat_processes',return_value=[proc]),patch('wechat_tool.mac_keys.signing_flags',return_value={'hardened_runtime':True,'get_task_allow':False}):
|
|
diagnostic=permission_diagnostic()
|
|
self.assertFalse(diagnostic['attempted'])
|
|
self.assertIsNone(diagnostic['task_for_pid_code'])
|
|
guidance=next_step(diagnostic)
|
|
self.assertEqual(guidance['action'],'capture')
|
|
self.assertEqual(guidance['level'],'warning')
|
|
self.assertIn('尚未实际读取',guidance['title'])
|
|
|
|
def test_denied_read_guides_to_admin_then_copy_without_auto_execution(self):
|
|
diagnostic={'code':'target_restricted','task_for_pid_code':5,'administrator':False}
|
|
self.assertEqual(next_step(diagnostic)['action'],'capture_admin')
|
|
diagnostic['administrator']=True
|
|
self.assertEqual(next_step(diagnostic)['action'],'preview_copy')
|
|
self.assertEqual(next_step(diagnostic,copy_ready=True)['action'],'launch_copy')
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
service=MacConnection(directory)
|
|
service.diagnostic=diagnostic
|
|
state=service.status()
|
|
self.assertEqual(state['next_step']['action'],'preview_copy')
|
|
self.assertFalse(list(Path(directory).rglob('*')))
|
|
self.assertFalse(service.recovery.plans)
|
|
|
|
def test_completed_check_does_not_claim_keys_or_import_succeeded(self):
|
|
diagnostic={'code':'target_restricted','task_for_pid_code':None,'administrator':False,
|
|
'summary':'只完成签名检查','attempted':False}
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
service=MacConnection(directory);service.job={'id':'check','action':'permissions','status':'running'}
|
|
with patch('wechat_tool.mac_connect.permission_diagnostic',return_value=diagnostic):
|
|
service.work('check','permissions',{},None,[])
|
|
self.assertEqual(service.job['status'],'complete')
|
|
self.assertEqual(service.job['stage'],'检查完成，等待选择读取方式')
|
|
self.assertEqual(service.job['result']['level'],'warning')
|
|
self.assertFalse(service.keys)
|
|
self.assertEqual(service.status()['next_step']['action'],'capture')
|
|
self.assertFalse(list(Path(directory).rglob('*')))
|
|
|
|
def test_other_checks_have_specific_manual_next_steps(self):
|
|
self.assertEqual(next_step({'code':'no_process'})['action'],'permissions')
|
|
self.assertEqual(next_step({'code':'multiple_processes'})['action'],'permissions')
|
|
self.assertEqual(next_step({'code':'can_attempt'})['action'],'capture')
|
|
self.assertEqual(next_step({'code':'access_denied','administrator':True,'task_for_pid_code':5})['action'],'key_file')
|
|
self.assertIsNone(next_step(None))
|
|
|
|
def test_administrator_helper_preserves_structured_failure(self):
|
|
diagnostic={'code':'target_restricted','summary':'目标签名拒绝调试','task_for_pid_code':5}
|
|
output=json.dumps({'error':'目标签名拒绝调试','diagnostic':diagnostic})
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
service=MacConnection(directory);service.job={'status':'running'}
|
|
with patch('wechat_tool.mac_connect.subprocess.run',return_value=subprocess.CompletedProcess([],0,output,'')):
|
|
service.work('id','capture',{'administrator':True},None,[])
|
|
self.assertEqual(service.job['status'],'failed')
|
|
self.assertEqual(service.job['error_code'],'target_restricted')
|
|
self.assertEqual(service.status()['diagnostic'],diagnostic)
|
|
|
|
|
|
class RecoveryTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp=tempfile.TemporaryDirectory()
|
|
self.base=Path(self.temp.name)
|
|
self.app=self.base/'Applications/WeChat.app'
|
|
for directory in ('Contents/MacOS','Contents/_CodeSignature'):
|
|
(self.app/directory).mkdir(parents=True)
|
|
(self.app/'Contents/Info.plist').write_bytes(plistlib.dumps({'CFBundleIdentifier':'com.tencent.xinWeChat','CFBundleExecutable':'WeChat','CFBundleShortVersionString':'4.1.0'}))
|
|
(self.app/'Contents/MacOS/WeChat').write_bytes(b'fake-app-binary')
|
|
(self.app/'Contents/_CodeSignature/CodeResources').write_bytes(b'fake-resource-seal')
|
|
self.account=self.base/'xwechat_files/account/db_storage'
|
|
self.account.mkdir(parents=True)
|
|
(self.account/'test.db').write_bytes(b'account-original')
|
|
all_users=self.account.parent.parent/'all_users'
|
|
all_users.mkdir()
|
|
(all_users/'login-config').write_bytes(b'login-original')
|
|
self.recovery=DebugRecovery(self.base/'workspace')
|
|
self.command_patch=patch('wechat_tool.mac_recovery.run',return_value=subprocess.CompletedProcess([],0,b'',b''))
|
|
self.runner=self.command_patch.start()
|
|
|
|
def tearDown(self):
|
|
self.command_patch.stop()
|
|
self.temp.cleanup()
|
|
|
|
def plan(self):
|
|
return self.recovery.plan(self.account,app=self.app)
|
|
|
|
def test_preview_is_readonly_and_consent_is_required(self):
|
|
result=self.plan()
|
|
self.assertFalse((self.base/'workspace').exists())
|
|
self.assertTrue(result['effects'])
|
|
with self.assertRaisesRegex(ValueError,'明确勾选'):
|
|
self.recovery.consume(result['plan_id'],False,self.account)
|
|
self.assertTrue(all('--force' not in call.args[0] for call in self.runner.call_args_list))
|
|
self.recovery.consume(result['plan_id'],True,self.account)
|
|
with self.assertRaisesRegex(ValueError,'已使用'):
|
|
self.recovery.consume(result['plan_id'],True,self.account)
|
|
|
|
def test_preparation_backs_up_before_signing_copy_and_preserves_original(self):
|
|
result=self.plan()
|
|
plan=self.recovery.consume(result['plan_id'],True,self.account)
|
|
original={str(p.relative_to(self.app)):p.read_bytes() for p in self.app.rglob('*') if p.is_file()}
|
|
def grants(app):
|
|
return {'com.apple.security.get-task-allow':app!=self.app}
|
|
def command(args,timeout=90):
|
|
if '--force' in args:
|
|
self.assertEqual(Path(args[-1]),plan['root']/'WeChat.app')
|
|
self.assertEqual((plan['root']/'backup/0-account/db_storage/test.db').read_bytes(),b'account-original')
|
|
self.assertEqual((plan['root']/'backup/1-all_users/login-config').read_bytes(),b'login-original')
|
|
return subprocess.CompletedProcess(args,0,b'',b'')
|
|
self.runner.side_effect=command
|
|
with patch('wechat_tool.mac_recovery.wechat_pid',return_value=None),patch('wechat_tool.mac_recovery.entitlements',side_effect=grants):
|
|
prepared=self.recovery.prepare(plan,lambda _:None)
|
|
self.assertTrue(prepared['original_unchanged'])
|
|
self.assertEqual(original,{str(p.relative_to(self.app)):p.read_bytes() for p in self.app.rglob('*') if p.is_file()})
|
|
mutations=[c.args[0] for c in self.runner.call_args_list if '--force' in c.args[0]]
|
|
self.assertEqual(len(mutations),1)
|
|
self.assertNotIn(str(self.app),mutations[0])
|
|
with patch('wechat_tool.mac_recovery.wechat_pid',return_value=None):
|
|
self.recovery.launch()
|
|
self.assertEqual(self.runner.call_args.args[0],['/usr/bin/open','-n',prepared['copy']])
|
|
|
|
def test_running_wechat_or_changed_original_prevents_preparation(self):
|
|
result=self.plan();plan=self.recovery.consume(result['plan_id'],True,self.account)
|
|
with patch('wechat_tool.mac_recovery.wechat_pid',return_value=123):
|
|
with self.assertRaisesRegex(ValueError,'退出所有微信'):
|
|
self.recovery.prepare(plan,lambda _:None)
|
|
(self.app/'Contents/MacOS/WeChat').write_bytes(b'changed')
|
|
with patch('wechat_tool.mac_recovery.wechat_pid',return_value=None):
|
|
with self.assertRaisesRegex(ValueError,'发生变化'):
|
|
self.recovery.prepare(plan,lambda _:None)
|
|
self.assertIsNone(self.recovery.prepared)
|
|
|
|
def test_unknown_or_tampered_copy_cannot_be_launched(self):
|
|
with self.assertRaisesRegex(ValueError,'先预览'):
|
|
self.recovery.launch()
|
|
self.recovery.prepared={'app':self.app,'fingerprint':'wrong'}
|
|
with patch('wechat_tool.mac_recovery.wechat_pid',return_value=None):
|
|
with self.assertRaisesRegex(ValueError,'已变化'):
|
|
self.recovery.launch()
|
|
|
|
def test_account_switch_and_expired_plan_are_rejected(self):
|
|
result=self.plan()
|
|
with self.assertRaisesRegex(ValueError,'账号发生变化'):
|
|
self.recovery.consume(result['plan_id'],True,self.account/'other')
|
|
result=self.plan()
|
|
self.recovery.plans[result['plan_id']]['created']-=1801
|
|
with self.assertRaisesRegex(ValueError,'过期'):
|
|
self.recovery.consume(result['plan_id'],True,self.account)
|