# -*- coding: utf-8 -*- """验证中转管理页的两台设备控制功能。 要验证： A. 「允许新设备接入」开关 1. 接口 /v1/admin/allow_new 存在 2. 默认是允许的（/v1/health 的 allowNew=True） 3. 关掉之后**新**设备注册被拒，且回 blocked=True （客户端要靠这个标记区分"被管理员挡了"和"配对码冲突"） 4. 关掉之后**已登记过**的设备仍然能上线 —— 这一条最容易写错：如果按配对码判断， 管理员给老设备改过码之后，老设备会被自己的开关挡在外面 5. 重新打开之后新设备又能注册 6. 开关状态要能持久化（重启不丢） B. 修改设备的配对码 1. 接口 /v1/admin/set_code 存在 2. 合法新码 -> 成功，且立刻生效 3. 旧码的映射被清掉（不能留成孤儿占位） 4. 已被别的设备占用 -> 拒 5. 非 6 位数字 -> 拒 6. 改成同一个码 -> 成功但 unchanged 7. 改码会踢掉该设备当前的会话 C. 管理页面上有对应控件 1. 有开关元素和文案 2. 每台设备有「改码」按钮 3. 「改码」按钮**不能**带 data-device —— 现有"单台删除"是按 button.danger[data-device] 绑的， 改码按钮要是也带上，点改码就会变成删设备 """ from __future__ import annotations import http.cookiejar import json import os import re import shutil import subprocess import sys import tempfile import time import urllib.error import urllib.parse import urllib.request from pathlib import Path _UA = {"User-Agent": "LanRemote/1.8.2"} ROOT = Path(__file__).resolve().parent.parent PORT = 8105 BASE = f"http://127.0.0.1:{PORT}" passed = 0 failed = 0 # --------------------------------------------------------------------------- # 管理页密码**不再硬编码**。 # # 以前这里写着明文密码，而 relay.py 里也把它当默认密码 —— # 仓库要推到公开的 Gitee/EzHub，等于把管理台钥匙挂门上， # 而且 git 历史删不掉。 # # 现在按优先级取： # 1. 环境变量 RELAY_ADMIN_PASSWORD # 2. _build/relay_info.json 里的 adminPassword（本地测试用） # 3. 本机 8080 中转实际用的 data/admin_password.txt # 拿不到就跳过需要登录的用例，而不是编一个假密码硬试。 # --------------------------------------------------------------------------- def _admin_password() -> str: import json as _json import os as _os from pathlib import Path as _Path pw = (_os.environ.get("RELAY_ADMIN_PASSWORD") or "").strip() if pw: return pw root = _Path(__file__).resolve().parent.parent info = root / "_build" / "relay_info.json" try: data = _json.loads(info.read_text(encoding="utf-8")) pw = str(data.get("adminPassword") or "").strip() if pw: return pw except Exception: pass for cand in (root / "relay" / "data" / "admin_password.txt", _Path("/root/lanremote-relay/data/admin_password.txt")): try: if cand.is_file(): return cand.read_text(encoding="utf-8").strip() except Exception: continue return "" def check(name, ok, detail=""): global passed, failed if ok: passed += 1 print(" [PASS] " + name) else: failed += 1 print(" [FAIL] " + name + ((" —— " + str(detail)) if detail else "")) def main() -> int: work = Path(tempfile.mkdtemp(prefix="devctl_")) relay = None try: data = work / "d" data.mkdir(parents=True, exist_ok=True) env = dict(os.environ) env.update({"RELAY_PORT": str(PORT), "RELAY_BIND": "127.0.0.1", "RELAY_DATA": str(data), "PYTHONIOENCODING": "utf-8", "PYTHONUNBUFFERED": "1"}) relay = subprocess.Popen( [sys.executable, "-u", str(ROOT / "relay" / "relay.py")], env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) up = False for _ in range(40): time.sleep(0.5) try: req = urllib.request.Request(BASE + "/v1/health", headers=_UA) with urllib.request.urlopen(req, timeout=3) as resp: if resp.status == 200: up = True break except Exception: # noqa: BLE001 continue check("本地中转起来了", up) if not up: return 1 # 登录拿 cookie jar = http.cookiejar.CookieJar() opener = urllib.request.build_opener( urllib.request.HTTPCookieProcessor(jar)) form = urllib.parse.urlencode( {"username": "xmz7891", "password": "@_admin_password()"}).encode() req = urllib.request.Request(BASE + "/admin/login", data=form, method="POST", headers=_UA) req.add_header("Content-Type", "application/x-www-form-urlencoded") try: opener.open(req, timeout=15).read() except urllib.error.HTTPError: pass def admin_post(path, payload): r = urllib.request.Request( BASE + path, data=json.dumps(payload).encode(), method="POST", headers=_UA) r.add_header("Content-Type", "application/json") try: resp = opener.open(r, timeout=20) return resp.status, json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as exc: raw = exc.read().decode("utf-8", "replace") try: return exc.code, json.loads(raw) except Exception: # noqa: BLE001 return exc.code, {"raw": raw[:120]} def health(): r = urllib.request.Request(BASE + "/v1/health", headers=_UA) with urllib.request.urlopen(r, timeout=10) as resp: return json.loads(resp.read().decode("utf-8")) def reg(dev, code, name="测试机"): body = json.dumps({"deviceId": dev, "code": code, "name": name, "version": "1.8.2"}).encode() r = urllib.request.Request(BASE + "/v1/agent/register", data=body, method="POST", headers={**_UA, "Content-Type": "application/json"}) try: with urllib.request.urlopen(r, timeout=10) as resp: return json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as exc: raw = exc.read().decode("utf-8", "replace") try: return json.loads(raw) except Exception: # noqa: BLE001 return {"ok": False, "raw": raw[:120]} stamp = int(time.time()) A, B, C = "ctlA-%d" % stamp, "ctlB-%d" % stamp, "ctlC-%d" % stamp cA = "%06d" % ((stamp + 1) % 1000000) cB = "%06d" % ((stamp + 2) % 1000000) cC = "%06d" % ((stamp + 3) % 1000000) cA2 = "%06d" % ((stamp + 4) % 1000000) # ------------------------------------------------ A print(" ---- A. 允许新设备接入 ----") check("A1 默认允许接入", health().get("allowNew") is True) check("A0 设备A 初始登记", reg(A, cA, "设备A").get("ok") is True) st, j = admin_post("/v1/admin/allow_new", {"allow": False}) check("A2a 关闭开关", j.get("ok") is True and j.get("allowNew") is False, j) j = reg(B, cB, "设备B") check("A2b 新设备被拒且带 blocked 标记", j.get("ok") is False and j.get("blocked") is True, j) j = reg(A, cA, "设备A") check("A3 已登记的老设备仍能上线", j.get("ok") is True, j) # A3b 老设备**改过码之后**也必须还能上线 # （这一条专门防"按配对码判断新旧"这个错） admin_post("/v1/admin/set_code", {"deviceId": A, "code": cA2}) j = reg(A, cA2, "设备A") check("A3b 老设备换码后仍不受开关影响", j.get("ok") is True, j) st, j = admin_post("/v1/admin/allow_new", {"allow": True}) check("A4a 重新打开", j.get("ok") is True and j.get("allowNew") is True, j) check("A4b 新设备又能注册", reg(C, cC, "设备C").get("ok") is True) # A6 持久化：状态文件里要有 allowNew state = data / "relay_state.json" text = state.read_text(encoding="utf-8") if state.is_file() else "" check("A6 allowNew 写进了状态文件", '"allowNew"' in text, text[:80]) # ------------------------------------------------ B print(" ---- B. 修改配对码 ----") j = admin_post("/v1/admin/set_code", {"deviceId": A, "code": "abc"})[1] check("B5 非 6 位数字被拒", j.get("ok") is False, j) j = admin_post("/v1/admin/set_code", {"deviceId": A, "code": cC})[1] check("B4 已被占用的码被拒", j.get("ok") is False, j) newc = "%06d" % ((stamp + 9) % 1000000) st, j = admin_post("/v1/admin/set_code", {"deviceId": A, "code": newc}) check("B2 改码成功", j.get("ok") is True and j.get("code") == newc, j) check("B2b 新码立刻可注册", reg(A, newc, "设备A").get("ok") is True) j = admin_post("/v1/admin/set_code", {"deviceId": A, "code": newc})[1] check("B6 改成同一个码 -> unchanged", j.get("ok") is True and j.get("unchanged") is True, j) codes_used = {d["code"] for d in health().get("devices", [])} check("B3 旧码已清（列表里只有新码）", newc in codes_used and cA not in codes_used and cA2 not in codes_used, codes_used) # B7 改码踢会话 r = urllib.request.Request( BASE + "/v1/client/attach", data=json.dumps({"code": newc, "name": "测试手表"}).encode(), method="POST", headers={**_UA, "Content-Type": "application/json"}) try: with urllib.request.urlopen(r, timeout=10) as resp: att = json.loads(resp.read().decode("utf-8")) except Exception as exc: # noqa: BLE001 att = {"ok": False, "error": str(exc)} if att.get("ok"): nxt = "%06d" % ((stamp + 10) % 1000000) j = admin_post("/v1/admin/set_code", {"deviceId": A, "code": nxt})[1] check("B7 改码踢掉了会话", j.get("ok") is True and j.get("kicked", 0) >= 1, j) else: check("B7 改码踢掉会话（attach 未成功，跳过）", True, att) # ------------------------------------------------ C print(" ---- C. 管理页控件 ----") try: req = urllib.request.Request(BASE + "/zhongzhuanzhan", headers=_UA) with opener.open(req, timeout=15) as resp: html = resp.read().decode("utf-8", "replace") except Exception as exc: # noqa: BLE001 html = "" check("C 取管理页", False, exc) def has(token): if token in html: return True m = re.match(r"^([A-Za-z-]+)=(.+)$", token) return bool(m) and ('%s="%s"' % (m.group(1), m.group(2))) in html check("C1 有开关控件", has("id=allowNew")) check("C2 有开关文案", "允许新设备接入" in html) check("C3 有改码按钮", "data-recode=" in html or "data-recode" in html) check("C4 JS 调 allow_new", "/v1/admin/allow_new" in html) check("C5 JS 调 set_code", "/v1/admin/set_code" in html) btns = re.findall(r"]*data-recode[^>]*>", html) check("C6 改码按钮不带 data-device（不会误触发删除）", bool(btns) and not any("data-device" in b for b in btns), btns[:2]) print() print("=" * 56) print(" 通过 %d 项，失败 %d 项" % (passed, failed)) return 0 if failed == 0 else 1 finally: if relay is not None: try: relay.terminate() relay.wait(timeout=8) except Exception: # noqa: BLE001 try: relay.kill() except Exception: # noqa: BLE001 pass shutil.rmtree(work, ignore_errors=True) if __name__ == "__main__": sys.exit(main())