# -*- coding: utf-8 -*- """验证静默模式。 三件事必须成立： 1. 启动后**一个窗口都没有** —— 用 EnumWindows 枚举该进程的所有顶层窗口来证明， 不是"看起来没有" 2. 真的在监听，手表能连上 3. **配对认证仍然生效** —— 静默模式默认不接受新设备， 否则局域网里任何人都能接进来，那就成了后门 """ from __future__ import annotations import ctypes import json import shutil import socket import subprocess import sys import tempfile import time from pathlib import Path ROOT = Path(__file__).resolve().parent.parent AGENT = ROOT / "windows" / "run_agent.py" DATA = ROOT / "windows" / "data" PORT = 8877 PASSED = 0 FAILED = 0 def check(name, ok, detail=""): global PASSED, FAILED if ok: PASSED += 1 print(f" [PASS] {name}") else: FAILED += 1 print(f" [FAIL] {name} {detail}") def windows_of_process(pid: int) -> list: """枚举属于该进程的所有可见/不可见顶层窗口。""" user32 = ctypes.windll.user32 found: list = [] EnumProc = ctypes.WINFUNCTYPE(ctypes.c_bool, ctypes.c_void_p, ctypes.c_void_p) def callback(hwnd, _lparam): owner = ctypes.c_ulong() user32.GetWindowThreadProcessId(hwnd, ctypes.byref(owner)) if owner.value == pid: length = user32.GetWindowTextLengthW(hwnd) buffer = ctypes.create_unicode_buffer(max(2, length + 1)) user32.GetWindowTextW(hwnd, buffer, length + 1) cls = ctypes.create_unicode_buffer(256) user32.GetClassNameW(hwnd, cls, 256) found.append({ "title": buffer.value, "class": cls.value, "visible": bool(user32.IsWindowVisible(hwnd)), }) return True user32.EnumWindows(EnumProc(callback), 0) return found def port_open(port: int) -> bool: try: sock = socket.create_connection(("127.0.0.1", port), timeout=1) sock.close() return True except OSError: return False def seed_paired_device(data_dir): """预置一台已配设备。 现在的规则是：**一台设备都没配对过时，配对是开着的** （这是修用户报的"设完码却连不上"）。所以想测 "静默模式拒绝新配对"，得先让电脑认为已经有设备配过了。 """ import json as _json data_dir = Path(data_dir) data_dir.mkdir(parents=True, exist_ok=True) (data_dir / "devices.json").write_text( _json.dumps({"already-paired": { "id": "already-paired", "name": "已配的手表", "token": "seed-token", "last": 0}}, ensure_ascii=False), encoding="utf-8") def seed_setup_done(data_dir): """预先置 setup_done：这些测试测的是别的功能，不该弹出首次设置窗口。""" import json as _json data_dir.mkdir(parents=True, exist_ok=True) settings = data_dir / "settings.json" payload = {} if settings.is_file(): try: payload = _json.loads(settings.read_text(encoding="utf-8")) except Exception: payload = {} payload["setup_done"] = True settings.write_text(_json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") def main() -> int: backup = None if DATA.exists(): backup = DATA.parent / f"data_backup_silent_{int(time.time())}" shutil.move(str(DATA), str(backup)) print(f" 已备份真实 data") seed_setup_done(DATA) seed_paired_device(DATA) proc = None try: proc = subprocess.Popen( [sys.executable, str(AGENT), "--port", str(PORT), "--silent"], stdout=subprocess.PIPE, stderr=subprocess.PIPE) print(f" 已静默启动（PID {proc.pid}），等待监听…") listening = False for _ in range(30): time.sleep(0.5) if port_open(PORT): listening = True break check("静默模式起来了并在监听", listening) print() print("一、窗口检查（EnumWindows 枚举该进程的顶层窗口）") time.sleep(1.5) windows = windows_of_process(proc.pid) visible = [w for w in windows if w["visible"]] # Tk 会创建一个隐藏的辅助窗口，那不算"界面窗口" ui_windows = [w for w in visible if w["title"]] print(f" 该进程顶层窗口 {len(windows)} 个，" f"可见 {len(visible)} 个，可见且有标题 {len(ui_windows)} 个") for w in windows[:6]: print(f" class={w['class']!r} title={w['title']!r} " f"visible={w['visible']}") check("没有任何可见窗口", not visible, f"竟然有 {len(visible)} 个: {visible[:3]}") check("没有带标题的界面窗口", not ui_windows, str(ui_windows[:3])) print() print("二、终端检查") check("启动时没有附带控制台窗口", not any(w["class"] == "ConsoleWindowClass" for w in windows)) print() print("三、★ 配对认证必须仍然生效") settings = DATA / "settings.json" code = "" if settings.is_file(): try: code = json.loads(settings.read_text(encoding="utf-8")).get( "pairing_code", "") except Exception: # noqa: BLE001 pass print(f" 配对码 {code or '(无)'}，静默模式默认不接受新设备") sock = socket.create_connection(("127.0.0.1", PORT), timeout=5) sock.settimeout(8) request = {"cmd": "pair", "deviceId": "silent-test", "deviceName": "测试", "code": code} sock.sendall((json.dumps(request, ensure_ascii=False) + "\n").encode("utf-8")) raw = b"" while b"\n" not in raw: chunk = sock.recv(4096) if not chunk: break raw += chunk reply = json.loads(raw.split(b"\n")[0].decode("utf-8")) if raw else {} print(f" 配对回复: {reply}") check("静默模式下新设备配对被拒绝", reply.get("ok") is False, f"竟然配上了: {reply}") check("给出的理由是明确的", "配对" in str(reply.get("error", "")) or "关闭" in str(reply.get("error", "")), str(reply.get("error"))) print() print("四、未配对的设备不能执行任何命令") sock.sendall((json.dumps({"cmd": "feature_list"}, ensure_ascii=False) + "\n").encode("utf-8")) raw = b"" while b"\n" not in raw: chunk = sock.recv(4096) if not chunk: break raw += chunk reply = json.loads(raw.split(b"\n")[0].decode("utf-8")) if raw else {} check("未配对时功能列表被拒绝", reply.get("ok") is False, f"竟然返回了: {str(reply)[:80]}") sock.close() print() print("五、日志文件") log = DATA / "silent.log" check("静默模式写了日志", log.is_file(), str(log)) finally: if proc is not None: try: proc.terminate() proc.wait(timeout=8) except Exception: # noqa: BLE001 try: proc.kill() except Exception: # noqa: BLE001 pass time.sleep(1.0) try: if DATA.exists(): shutil.rmtree(DATA, ignore_errors=True) if backup is not None and backup.exists(): shutil.move(str(backup), str(DATA)) print(" 已还原真实 data") except Exception as exc: # noqa: BLE001 print(f" !! 还原失败: {exc}") print() print("=" * 62) print(f"通过 {PASSED} 项，失败 {FAILED} 项") return 1 if FAILED else 0 if __name__ == "__main__": sys.exit(main())