# -*- coding: utf-8 -*- r'''验证假设：bench_lan_vs_relay 中转那半失败，是因为两条路 用了**同一个配对码**，第二台被控端注册被"一码一机"规则拒掉， 而 start_agent 看到上一台还挂在线就以为成功了。 做法：完全照 bench 的流程走两遍（起 agent1 -> 停 -> 起 agent2）， 第一遍用同一个码，第二遍用不同的码，看结果差异。 ''' from __future__ import annotations import json import os import shutil import socket import subprocess import sys import tempfile import time import urllib.error import urllib.request from pathlib import Path ROOT = Path(__file__).resolve().parent.parent WINDOWS = ROOT / "windows" UA = {"User-Agent": "LanRemote/1.8.2"} def free_port() -> int: s = socket.socket() s.bind(("127.0.0.1", 0)) p = s.getsockname()[1] s.close() return p def get(base, path, timeout=6.0): req = urllib.request.Request(base + path, headers=UA) try: with urllib.request.urlopen(req, timeout=timeout) as resp: return resp.status, resp.read() except Exception: # noqa: BLE001 return 0, b"" def post(base, path, body=b"", headers=None, timeout=20.0): h = dict(UA) h.update(headers or {}) req = urllib.request.Request(base + path, data=body, headers=h, method="POST") try: with urllib.request.urlopen(req, timeout=timeout) as resp: return resp.status, resp.read() except urllib.error.HTTPError as exc: return exc.code, exc.read() except Exception as exc: # noqa: BLE001 return 0, str(exc).encode() def main() -> int: work = Path(tempfile.mkdtemp(prefix="codechk_")) relay_port = free_port() agent_port = free_port() base = "http://127.0.0.1:%d" % relay_port relay_proc = None procs = [] try: sys.path.insert(0, str(WINDOWS)) from lanremote.config import Store rdata = work / "r" rdata.mkdir(parents=True, exist_ok=True) env = dict(os.environ) env.update({"RELAY_PORT": str(relay_port), "RELAY_BIND": "127.0.0.1", "RELAY_DATA": str(rdata), "RELAY_POLL": "2", "PYTHONIOENCODING": "utf-8", "PYTHONUNBUFFERED": "1"}) rlog = open(work / "relay.log", "wb") relay_proc = subprocess.Popen( [sys.executable, "-u", str(ROOT / "relay" / "relay.py")], env=env, stdout=rlog, stderr=subprocess.STDOUT) for _ in range(60): time.sleep(0.5) if get(base, "/v1/health")[0] == 200: break def start_agent(code, tag): d = work / tag d.mkdir(parents=True, exist_ok=True) st = Store(d) st.set_pairing_code(code) st.set("setup_done", True) st.set("allow_new_pairing", True) st.set("close_pair_after_first", False) st.set("server_name", "验码电脑") st.set("relay_url", base) st.set("port", agent_port) e = dict(os.environ) e.update({"YKXMZ_DATA": str(d), "YKXMZ_LOCK_PORT": str(free_port()), "PYTHONIOENCODING": "utf-8", "PYTHONUNBUFFERED": "1"}) p = subprocess.Popen( [sys.executable, "-u", str(WINDOWS / "run_agent.py"), "--silent", "--port", str(agent_port)], cwd=str(WINDOWS), env=e, stdout=open(d / "out.log", "wb"), stderr=subprocess.STDOUT) procs.append(p) online = False for _ in range(30): time.sleep(1.0) s, b = get(base, "/v1/health", timeout=4) if s == 200: snap = json.loads(b.decode()) if any(x.get("online") for x in snap.get("devices", [])): online = True break return p, online, snap if s == 200 else {} def stop(p): try: p.terminate() p.wait(timeout=6) except Exception: # noqa: BLE001 try: p.kill() except Exception: # noqa: BLE001 pass def relay_try(code, label): """attach + pair，看能不能拿到回复。""" s, b = post(base, "/v1/client/attach", json.dumps( {"code": code, "name": "验码手表"}).encode(), {"Content-Type": "application/json"}) att = json.loads(b.decode()) if not att.get("ok"): print(" %s attach 就失败: %s" % (label, att.get("error"))) return False sess = att.get("session", "") post(base, "/v1/client/send", (json.dumps({"cmd": "pair", "deviceId": "x", "code": code}) + "\n").encode(), {"X-Session": sess}) buf = b"" deadline = time.time() + 12 while time.time() < deadline: _s, chunk = post(base, "/v1/client/poll", b"", {"X-Session": sess}, 6.0) buf += chunk while b"\n" in buf: line, _, buf = buf.partition(b"\n") if not line.strip(): continue msg = json.loads(line.decode()) if "event" in msg: continue print(" %s pair 成功: %s" % (label, json.dumps(msg, ensure_ascii=False)[:80])) return True print(" %s pair **超时**（没人回包）" % label) return False # ============ 场景 A：两条路用同一个码（bench 现在的做法）============ print("════════ 场景 A：两条路用同一个配对码（复刻 bench 现状）════════") CODE = "246813" p1, on1, snap1 = start_agent(CODE, "a1") print(" 第 1 台（跑局域网）: 登记=%s 设备数=%d" % (on1, len(snap1.get("devices", [])))) stop(p1) print(" 已停掉第 1 台") time.sleep(2) p2, on2, snap2 = start_agent(CODE, "a2") print(" 第 2 台（跑中转）: start_agent 认为登记=%s" % on2) for d in snap2.get("devices", []): print(" 中转上看到的设备: %s 码 %s 在线=%s 空闲=%ss" % (d["name"], d["code"], d["online"], d["idleSeconds"])) okA = relay_try(CODE, "场景A") stop(p2) print() print(" ── 中转日志里有没有『码被占用』──") rlog.flush() for line in (work / "relay.log").read_text( encoding="utf-8", errors="replace").splitlines(): if "占用" in line or "拒绝" in line: print(" " + line[:110]) # ============ 场景 B：每条路用不同的码 ============ print() print("════════ 场景 B：两条路各用不同的配对码（修法）════════") work2 = work / "b" CODE1 = "111222" CODE2 = "333444" p3, on3, _ = start_agent(CODE1, "b1") print(" 第 1 台: 登记=%s" % on3) stop(p3) time.sleep(2) p4, on4, snap4 = start_agent(CODE2, "b2") print(" 第 2 台: 登记=%s" % on4) okB = relay_try(CODE2, "场景B") stop(p4) print() print("════════ 结论 ════════") print(" 场景 A（同一个码）: %s" % ("通" if okA else "**不通**")) print(" 场景 B（不同码） : %s" % ("通" if okB else "**不通**")) if (not okA) and okB: print(" → 假设成立：bench 失败是因为两条路用了同一个配对码，") print(" 第 2 台被「一码一机」规则拒掉，而 start_agent") print(" 看到第 1 台还挂在线就以为成功了。**产品没问题。**") return 0 finally: for p in procs: try: p.terminate() except Exception: # noqa: BLE001 pass if relay_proc: try: relay_proc.terminate() relay_proc.wait(timeout=6) except Exception: # noqa: BLE001 pass shutil.rmtree(work, ignore_errors=True) if __name__ == "__main__": sys.exit(main())