#!/usr/bin/env python3 """工作手机执行前只读总门禁：一次检查高频重复问题。""" from __future__ import annotations import argparse import hashlib import json import subprocess import sys from datetime import datetime from pathlib import Path from typing import Any from urllib.error import URLError from urllib.request import ProxyHandler, build_opener ROOT = Path(__file__).resolve().parents[1] HOOK_MIRRORS = [ ROOT / "agent/hook/wechat_hook_v2.js", ROOT / "app/agent/hook/wechat_hook_v2.js", ROOT / "android-app/app/src/main/assets/wechat_hook_v2.js", ] WRITE_CONTRACTS = { "/api/v3/message/send": {"dry_run", "confirm", "idempotency_key", "trace_id"}, "/api/v3/moments/post": {"dry_run", "confirm", "idempotency_key", "trace_id"}, "/api/v3/wechat/friend-group/execute": { "dry_run", "confirm", "idempotency_key", "trace_id", }, "/api/v3/payment/transfer": { "dry_run", "confirm", "idempotency_key", "trace_id", }, } def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def schema_properties(openapi: dict[str, Any], path: str) -> set[str]: schema = ( openapi.get("paths", {}) .get(path, {}) .get("post", {}) .get("requestBody", {}) .get("content", {}) .get("application/json", {}) .get("schema", {}) ) ref = str(schema.get("$ref") or "") if ref: schema = openapi.get("components", {}).get("schemas", {}).get(ref.rsplit("/", 1)[-1], {}) return set((schema.get("properties") or {}).keys()) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--base", default="http://127.0.0.1:8899") parser.add_argument("--device-id", default="") parser.add_argument("--timeout", type=float, default=5) parser.add_argument("--output", type=Path) parser.add_argument("--skip-interface-audit", action="store_true") args = parser.parse_args() opener = build_opener(ProxyHandler({})) def get_json(path: str) -> dict[str, Any]: with opener.open(args.base.rstrip("/") + path, timeout=args.timeout) as response: return json.load(response) checks: list[dict[str, Any]] = [] def add(name: str, ok: bool, *, blocker: str = "", detail: Any = None) -> None: checks.append({"name": name, "ok": ok, "blocker": blocker if not ok else "", "detail": detail}) health: dict[str, Any] = {} devices: list[dict[str, Any]] = [] try: health = get_json("/health") add( "sdk_health", health.get("status") == "healthy", blocker="environment/sdk_unhealthy", detail={ "version": health.get("version"), "devices_online": health.get("devices_online"), "device_ids": health.get("device_ids"), }, ) except (OSError, URLError, ValueError, json.JSONDecodeError) as exc: add("sdk_health", False, blocker="environment/sdk_unreachable", detail=type(exc).__name__) try: payload = get_json("/api/v3/devices") devices = payload.get("data") if isinstance(payload.get("data"), list) else [] online = [item for item in devices if item.get("status") == "online"] add( "single_online_device", len(online) == 1, blocker="connection/device_identity_not_unique", detail=[item.get("device_id") for item in online], ) except (OSError, URLError, ValueError, json.JSONDecodeError) as exc: online = [] add("single_online_device", False, blocker="connection/device_list_failed", detail=type(exc).__name__) device_id = args.device_id or (str(online[0].get("device_id")) if len(online) == 1 else "") if device_id: try: hook = get_json(f"/api/v3/hook/probe/{device_id}") hook_ok = bool(hook.get("supports_hook") and hook.get("hook_available_reported")) add( "hook_rpc_ready", hook_ok, blocker="capability/frida_hook_not_attached", detail={ "device_id": device_id, "supports_hook": hook.get("supports_hook"), "hook_available_reported": hook.get("hook_available_reported"), "frida_available_reported": hook.get("frida_available_reported"), "transport": hook.get("transport"), "probe_detail": hook.get("probe_detail"), }, ) except (OSError, URLError, ValueError, json.JSONDecodeError) as exc: add("hook_rpc_ready", False, blocker="capability/hook_probe_failed", detail=type(exc).__name__) else: add("hook_rpc_ready", False, blocker="connection/no_unique_device", detail={}) hashes = {str(path.relative_to(ROOT)): sha256(path) for path in HOOK_MIRRORS if path.exists()} add( "hook_source_sha_consistent", len(hashes) == len(HOOK_MIRRORS) and len(set(hashes.values())) == 1, blocker="runtime/hook_mirror_sha_drift", detail=hashes, ) try: openapi = get_json("/openapi.json") missing: dict[str, list[str]] = {} for path, required in WRITE_CONTRACTS.items(): absent = sorted(required - schema_properties(openapi, path)) if absent: missing[path] = absent add( "write_contract_fields", not missing, blocker="contract/openapi_write_gate_fields_missing", detail=missing, ) except (OSError, URLError, ValueError, json.JSONDecodeError) as exc: add("write_contract_fields", False, blocker="contract/openapi_unavailable", detail=type(exc).__name__) if not args.skip_interface_audit: audit = subprocess.run( [sys.executable, str(ROOT / "scripts/wechat_interface_audit.py")], cwd=ROOT.parent, capture_output=True, text=True, timeout=30, ) add( "action_mapping_audit", audit.returncode == 0, blocker="capability/action_mapping_drift", detail=(audit.stdout + audit.stderr).strip().splitlines()[-8:], ) hard_read_checks = {"sdk_health", "single_online_device"} ready_for_read = all(item["ok"] for item in checks if item["name"] in hard_read_checks) ready_for_write = all(item["ok"] for item in checks) result = { "checked_at": datetime.now().astimezone().isoformat(), "sdk_base": args.base, "device_id": device_id, "ready_for_read": ready_for_read, "ready_for_write": ready_for_write, "blockers": [item["blocker"] for item in checks if not item["ok"]], "checks": checks, } rendered = json.dumps(result, ensure_ascii=False, indent=2) print(rendered) if args.output: args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(rendered + "\n", encoding="utf-8") return 0 if ready_for_write else 2 if __name__ == "__main__": raise SystemExit(main())