# -*- coding: utf-8 -*- """验证用户最在意的那条路径： 配一次 → 以后开机静默启动 → 手表自动连上 → **全程零窗口零弹窗** 分四步： 1. 带窗口启动，用手表配对（这一步是唯一需要交互的，只做一次） 2. 关掉，改成静默启动 3. 手表用令牌重连（不再需要配对码） 4. 一边连着一边枚举该进程的窗口 —— 必须一个都没有 """ 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 = 8881 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(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 class Talk: def __init__(self, port, timeout=8.0): self.sock = socket.create_connection(("127.0.0.1", port), timeout=timeout) self.sock.settimeout(timeout) self.buf = b"" def call(self, obj, timeout=10.0): self.sock.sendall((json.dumps(obj, ensure_ascii=False) + "\n").encode("utf-8")) deadline = time.time() + timeout while True: while b"\n" not in self.buf: if time.time() > deadline: raise TimeoutError("超时") chunk = self.sock.recv(65536) if not chunk: raise ConnectionError("断开") self.buf += chunk line, _, self.buf = self.buf.partition(b"\n") message = json.loads(line.decode("utf-8")) # 服务端会主动推 event（剪贴板、画面帧…），那不是回复，要跳过 if "event" in message: continue return message def close(self): try: self.sock.close() except OSError: pass def wait_port(port, seconds=25, must_close=True): deadline = time.time() + seconds while time.time() < deadline: try: sock = socket.create_connection(("127.0.0.1", port), timeout=1) sock.close() return True except OSError: time.sleep(0.4) return False def stop_agent(proc): if proc is None: return try: proc.terminate() proc.wait(timeout=8) except Exception: # noqa: BLE001 try: proc.kill() except Exception: # noqa: BLE001 pass time.sleep(1.2) 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_bk_flow_{int(time.time())}" shutil.move(str(DATA), str(backup)) seed_setup_done(DATA) proc = None try: # ---------------- 第一步：带窗口配对一次 ---------------- print("一、第一次配对（唯一需要交互的一步）") # --add-device：静默运行 + 临时开放配对（等价于"第一次配对"那段） proc = subprocess.Popen([sys.executable, str(AGENT), "--port", str(PORT), "--add-device", "10"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) check("被控端起来了", wait_port(PORT)) settings = DATA / "settings.json" code = "" for _ in range(20): time.sleep(0.5) if settings.is_file(): try: code = json.loads(settings.read_text( encoding="utf-8")).get("pairing_code", "") except Exception: # noqa: BLE001 pass if code: break check("读到配对码", bool(code), code) talk = Talk(PORT) reply = talk.call({"cmd": "pair", "deviceId": "watch-001", "deviceName": "老师的手表", "code": code}) check("配对成功", reply.get("ok") is True, str(reply)[:100]) token = reply.get("token", "") check("拿到了令牌（以后靠它重连）", bool(token), str(reply)[:80]) talk.close() stop_agent(proc) proc = None # ---------------- 第二步：改成静默启动 ---------------- print() print("二、改成静默启动（模拟开机自启那条命令）") proc = subprocess.Popen([sys.executable, str(AGENT), "--port", str(PORT), "--silent"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) check("静默模式起来了", wait_port(PORT)) time.sleep(1.5) before = windows_of(proc.pid) check("启动时就没有任何窗口", not [w for w in before if w["visible"]], str([w for w in before if w["visible"]][:3])) # ---------------- 第三步：用令牌重连 ---------------- print() print("三、手表用令牌重连（不需要再输配对码）") talk = Talk(PORT) reply = talk.call({"cmd": "auth", "deviceId": "watch-001", "deviceName": "老师的手表", "token": token}) check("令牌认证通过", reply.get("ok") is True, str(reply)[:120]) # ---------------- 第四步：连着的时候也一个窗口都没有 ---------------- print() print("四、★ 连着的时候枚举窗口") time.sleep(1.5) after = windows_of(proc.pid) visible = [w for w in after if w["visible"]] print(f" 顶层窗口 {len(after)} 个，可见 {len(visible)} 个") for w in after[:5]: print(f" class={w['class']!r} title={w['title']!r} " f"visible={w['visible']}") check("★ 连上之后仍然没有任何窗口", not visible, str(visible[:3])) # 真的能用 reply = talk.call({"cmd": "feature_list"}, timeout=25) features = reply.get("features") or [] if not features or reply.get("ok") is not True: print(f" （feature_list 原始回复: {str(reply)[:300]}）") check("已配对设备能拿到功能列表", reply.get("ok") is True and len(features) > 100, f"{len(features)} 个, ok={reply.get('ok')}, " f"err={reply.get('error')}") reply = talk.call({"cmd": "perf"}, timeout=20) check("能执行命令", reply.get("ok") is True, str(reply)[:80]) # 未配对设备仍然进不来 print() print("五、未配对设备依旧被挡") intruder = Talk(PORT) # 用**错误的码** —— 安全边界是"码错必拒"。 # （以前这里用正确的码也期待被拒，但行为改成 # "配对码本身就是密钥，码对就放行"了。） wrong = "000000" if code != "000000" else "111111" bad = intruder.call({"cmd": "pair", "deviceId": "student-phone", "deviceName": "陌生设备", "code": wrong}) # 注意：这里以前是拿**正确的码**去试，期待被拒。 # 后来改成"码对就放行"（配对码本身就是密钥）， # 所以安全边界换成"**错误的码**必须被拒"。 check("拿着错误的码配不上", bad.get("ok") is False, str(bad)[:100]) intruder.close() # 表格重装/换新表：拿着正确的码应该能重新配上 again = Talk(PORT) retry = again.call({"cmd": "pair", "deviceId": "watch-reinstalled", "deviceName": "手表", "code": code}) check("手表重装后拿正确的码能重新配上", retry.get("ok") is True, str(retry)[:120]) again.close() talk.close() finally: stop_agent(proc) 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())