177 lines
8.4 KiB
Python
177 lines
8.4 KiB
Python
"""WP-HL-07 资料设置：内核接入前的独立契约门禁。"""
|
||
from pathlib import Path
|
||
|
||
from sdk.agent.hook.companion_dispatch import ACCOUNT_SETTING_CONTRACTS, dispatch_record
|
||
|
||
ROOT = Path(__file__).resolve().parents[2]
|
||
HOOK = ROOT / "sdk" / "agent" / "hook" / "wechat_hook_v2.js"
|
||
|
||
PROFILE_VECTORS = {
|
||
"set_nickname": {
|
||
"rpc": "setNickname", "params": {"nickname": "测试昵称-A"},
|
||
"readback_rpc": "getProfile", "readback": {"nickname": "测试昵称-A"},
|
||
},
|
||
"set_signature": {
|
||
"rpc": "setSignature", "params": {"signature": "测试签名-A"},
|
||
"readback_rpc": "getProfile", "readback": {"signature": "测试签名-A"},
|
||
},
|
||
"set_region": {
|
||
"rpc": "setRegion", "params": {"region": "福建 厦门"},
|
||
"readback_rpc": "getProfile", "readback": {"region": "福建 厦门"},
|
||
},
|
||
"set_sex": {
|
||
"rpc": "setSex", "params": {"sex": 1},
|
||
"readback_rpc": "getProfile", "readback": {"sex": 1},
|
||
},
|
||
"set_avatar": {
|
||
"rpc": "setAvatar", "params": {"image_path": "/sdcard/Download/wp-hl-07-avatar.jpg", "image_sha256": "a" * 64},
|
||
"readback_rpc": "getProfile", "readback": {"avatar": {"sha256": "a" * 64}},
|
||
},
|
||
"set_privacy": {
|
||
"rpc": "setPrivacy", "params": {"setting": "moments_visibility", "value": "all"},
|
||
"readback_rpc": "getPrivacy", "readback": {"setting": "moments_visibility", "value": "all"},
|
||
},
|
||
"set_account_protection": {
|
||
"rpc": "setAccountProtection", "params": {"enable": True},
|
||
"readback_rpc": "getAccountProtection", "readback": {"enable": True},
|
||
},
|
||
"set_notification": {
|
||
"rpc": "setNotification", "params": {"type": "message", "enable": True},
|
||
"readback_rpc": "getNotification", "readback": {"type": "message", "enable": True},
|
||
},
|
||
"set_do_not_disturb": {
|
||
"rpc": "setDoNotDisturb", "params": {"enable": True, "wxid": "wxid_test_contact"},
|
||
"readback_rpc": "getDoNotDisturb", "readback": {"enable": True, "wxid": "wxid_test_contact"},
|
||
},
|
||
"set_what_up": {
|
||
"rpc": "setWhatUp", "params": {"content": "测试状态-A"},
|
||
"readback_rpc": "getWhatUp", "readback": {"content": "测试状态-A"},
|
||
},
|
||
}
|
||
|
||
|
||
def test_profile_actions_match_dispatch_contract_and_readback_vectors():
|
||
for action, vector in PROFILE_VECTORS.items():
|
||
contract = ACCOUNT_SETTING_CONTRACTS[action]
|
||
dispatched = dispatch_record(action)
|
||
assert contract["rpc"] == vector["rpc"] == dispatched["rpc"]
|
||
assert set(contract["required"]).issubset(vector["params"])
|
||
assert dispatched["test_status"] == "contract_dry_run_pending_internal_rpc"
|
||
|
||
|
||
def test_profile_vectors_require_dry_run_confirm_and_three_evidence_fields():
|
||
for vector in PROFILE_VECTORS.values():
|
||
payload = {**vector["params"], "dry_run": False, "confirm": True}
|
||
assert payload["dry_run"] is False
|
||
assert payload["confirm"] is True
|
||
expected_evidence = {"trace_id", "channel", "raw_rpc_receipt", "readback"}
|
||
assert expected_evidence == {"trace_id", "channel", "raw_rpc_receipt", "readback"}
|
||
|
||
|
||
def test_profile_hook_exports_are_guarded_until_internal_write_signature_is_verified():
|
||
source = HOOK.read_text(encoding="utf-8")
|
||
for action, vector in PROFILE_VECTORS.items():
|
||
assert f"{action}: {{" in source
|
||
assert f"rpc: '{vector['rpc']}'" in source
|
||
assert f"readback_rpc: '{vector['readback_rpc']}'" in source
|
||
for field in vector["readback"]:
|
||
assert f"'{field}'" in source
|
||
assert "rpc.exports.probeAccountSettingAdapters" in source
|
||
assert "adapter_candidate_found_probe_only" in source
|
||
guard = source[source.index("function _accountSettingContract"):source.index("function _companionCapabilityUnavailable")]
|
||
assert "capability_unavailable" in guard
|
||
assert "当前版本尚无可验证内部写入调用策略" in guard
|
||
assert "_intentAction(" not in guard
|
||
|
||
|
||
def test_sex_alias_and_do_not_disturb_alias_remain_semantically_separate():
|
||
source = HOOK.read_text(encoding="utf-8")
|
||
assert "canonical: 'set_sex'" in source
|
||
assert "canonical: 'set_do_not_disturb'" in source
|
||
assert PROFILE_VECTORS["set_sex"]["params"] == {"sex": 1}
|
||
assert PROFILE_VECTORS["set_do_not_disturb"]["params"]["enable"] is True
|
||
|
||
READBACK_VECTORS_36_39 = {
|
||
"36_contacts": {
|
||
"sdk": "GET /api/v3/contacts",
|
||
"rpc": "getContacts",
|
||
"params": {"limit": 2, "offset": 0},
|
||
"required_contact": {"wxid", "user_id", "display_name", "type", "sex", "tags", "has_avatar"},
|
||
"identity": "wxid",
|
||
},
|
||
"37_profile_bundle": {
|
||
"sdk": "GET /api/v3/customer/profile-bundle",
|
||
"bff": "/v1/workphone/customer/profile-bundle",
|
||
"required_bundle": {"profile", "contacts", "contact_count", "returned_contact_count", "contact_total_count", "contacts_has_more", "channel_used"},
|
||
"identity": "device_id_md5+contact.wxid",
|
||
},
|
||
"38_current_profile": {
|
||
"sdk": "GET /api/v3/profile/get",
|
||
"rpc": "getProfile",
|
||
"required_profile": {"wxid", "nickname", "phone", "signature", "sex", "region", "wechat_version"},
|
||
"identity": "wxid",
|
||
},
|
||
"39_friend_detail": {
|
||
"sdk": "GET /api/v3/friend/info",
|
||
"rpc": "getContactInfo",
|
||
"params": {"user_id": "wxid_test_contact"},
|
||
"required_contact": {"username", "nickname", "conRemark", "type", "alias"},
|
||
"identity": "contact.username == requested user_id",
|
||
},
|
||
}
|
||
|
||
|
||
def test_36_39_readback_vectors_are_bound_to_wireless_frida_read_rpcs():
|
||
source = HOOK.read_text(encoding="utf-8")
|
||
assert "getContacts: function" in source
|
||
assert "getContactInfo: function" in source
|
||
assert "getProfile: function" in source
|
||
assert "function _getContactsFromDB(limit, offset)" in source
|
||
assert "function _getContactInfoFromDB(wxid)" in source
|
||
assert READBACK_VECTORS_36_39["36_contacts"]["rpc"] == "getContacts"
|
||
assert READBACK_VECTORS_36_39["38_current_profile"]["rpc"] == "getProfile"
|
||
assert READBACK_VECTORS_36_39["39_friend_detail"]["rpc"] == "getContactInfo"
|
||
|
||
|
||
def test_36_contacts_normalization_has_stable_bff_diff_identity_and_display_name():
|
||
normalizer = (ROOT / "sdk" / "app" / "services" / "wechat_contact_normalizer.py").read_text(encoding="utf-8")
|
||
hook = HOOK.read_text(encoding="utf-8")
|
||
for field in READBACK_VECTORS_36_39["36_contacts"]["required_contact"]:
|
||
assert f'"{field}"' in normalizer or f'{field}:' in hook
|
||
assert 'display = nickname or remark or alias or wxid' in normalizer
|
||
assert '"user_id": wxid' in normalizer
|
||
assert "contactLabelMap[wxid] || []" in hook
|
||
|
||
|
||
def test_37_profile_bundle_keeps_child_channels_counts_and_friend_detail_readback():
|
||
route = (ROOT / "sdk" / "app" / "routers" / "unified.py").read_text(encoding="utf-8")
|
||
start = route.index('async def get_customer_profile_bundle(')
|
||
block = route[start:route.index('@router.get("/friend/info"', start)]
|
||
for action in ("get_profile", "get_contacts", "get_groups", "get_tags", "get_messages"):
|
||
assert f'"{action}"' in block
|
||
for field in READBACK_VECTORS_36_39["37_profile_bundle"]["required_bundle"] - {"channel_used"}:
|
||
assert f'"{field}"' in block
|
||
assert 'channels[name] = result.get("_channel_used", "sdk_control")' in block
|
||
assert '"get_friend_info"' in block
|
||
|
||
|
||
def test_38_profile_and_39_friend_detail_require_exact_identity_readback():
|
||
source = HOOK.read_text(encoding="utf-8")
|
||
profile = source[source.index('getProfile: function'):source.index('checkAccountStatus: function')]
|
||
for field in READBACK_VECTORS_36_39["38_current_profile"]["required_profile"]:
|
||
assert f'{field}:' in profile
|
||
detail = source[source.index('function _getContactInfoFromDB(wxid)'):source.index('// ============================================================\n// § 7', source.index('function _getContactInfoFromDB(wxid)'))]
|
||
assert "WHERE username='" in detail
|
||
assert "requested_user_id: wxid" in detail
|
||
for field in READBACK_VECTORS_36_39["39_friend_detail"]["required_contact"]:
|
||
assert field in detail
|
||
|
||
|
||
def test_profile_writes_do_not_fall_back_to_device_frida_cli_when_server_bridge_is_available():
|
||
route = (ROOT / "sdk" / "app" / "routers" / "unified.py").read_text(encoding="utf-8")
|
||
start = route.index("async def _execute_skill(")
|
||
block = route[start:route.index("# 〇-B、Frida Hook", start)]
|
||
assert "account_write_actions" in block
|
||
assert "action in account_write_actions" in block
|
||
assert "不进入设备 Frida CLI/UI 回退" in block
|