ykxmz/_tests/test_verify_ttl_lockout.py

# -*- coding: utf-8 -*-
"""把第一个 bug 钉死：DEVICE_TTL 清理 + allow_new=False = 设备永久锁死。

要回答的问题：
  1. 现象能不能稳定复现？
  2. **重启中转能不能救回来？**
     这条最关键 ——
       snapshot 清 codes 时**没调 _save()**，所以磁盘上的 relay_state.json
       还留着那台设备的码绑定。重启后 _load() 会把 codes 恢复回来，
       但 devices 仍然是空的 -> known_device 仍然是 False -> 照样被挡。
     如果重启也救不回来，那就是"管理员必须手动打开开关"的死局。
  3. 真实场景下多久会触发？
     DEVICE_TTL 默认 6 小时。用户下班关机、第二天开机 = 超过 6 小时。
     也就是说**只要管理员关过这个开关，隔夜再开机的电脑就再也连不上**。
  4. 有没有别的入口能绕过（比如设备换个 deviceId 重新注册）？
     —— 换个 deviceId 就是"新设备"，同样被挡。
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
PORT = 8139
BASE = f"http://127.0.0.1:{PORT}"
_UA = {"User-Agent": "LanRemote/1.8.2"}
ADMIN_KEY = "ttl-lockout-key"
TTL = 2.0

passed = 0
failed = 0


def check(name: str, ok: bool, detail: str = "") -> bool:
    global passed, failed
    if ok:
        passed += 1
        print("  [PASS] " + name)
    else:
        failed += 1
        print("  [FAIL] " + name + (("  —— " + str(detail)) if detail else ""))
    return ok


def main() -> int:
    work = Path(tempfile.mkdtemp(prefix="ttl_lock_"))
    relay = None
    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), "RELAY_DEVICE_TTL": str(TTL),
        "RELAY_ADMIN_KEY": ADMIN_KEY, "RELAY_POLL": "2",
        "PYTHONIOENCODING": "utf-8", "PYTHONUNBUFFERED": "1",
    })

    def start() -> subprocess.Popen:
        p = subprocess.Popen(
            [sys.executable, "-u", str(ROOT / "relay" / "relay.py")], env=env,
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        for _ in range(60):
            time.sleep(0.5)
            try:
                health()
                return p
            except Exception:                              # noqa: BLE001
                continue
        return p

    def health() -> dict:
        r = urllib.request.Request(BASE + "/v1/health", headers=_UA)
        with urllib.request.urlopen(r, timeout=5) as resp:
            return json.loads(resp.read().decode("utf-8"))

    def admin(payload: dict) -> dict:
        body = json.dumps(payload).encode()
        r = urllib.request.Request(
            BASE + "/v1/admin/allow_new", data=body, method="POST",
            headers={**_UA, "X-Admin": ADMIN_KEY,
                     "Content-Type": "application/json"})
        try:
            with urllib.request.urlopen(r, timeout=15) as resp:
                return json.loads(resp.read().decode("utf-8"))
        except urllib.error.HTTPError as exc:
            try:
                return json.loads(exc.read().decode("utf-8"))
            except Exception:                              # noqa: BLE001
                return {}

    def reg(dev: str, code: str, name: str = "家里的电脑") -> dict:
        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:
            try:
                return json.loads(exc.read().decode("utf-8"))
            except Exception:                              # noqa: BLE001
                return {"ok": False}

    try:
        relay = start()

        print()
        print("═══ 复现：管理员关掉「允许新设备接入」，电脑隔夜再开机 ═══")
        dev = "home-pc-DESKTOP-ABC123"
        code = "246810"

        r1 = reg(dev, code)
        check("① 电脑首次登记成功", r1.get("ok") is True, r1)

        res = admin({"allow": False})
        check("② 管理员关掉「允许新设备接入」", res.get("allowNew") is False, res)

        r2 = reg(dev, code)
        check("③ 电脑当场重连仍然可以（开关只挡新设备）",
              r2.get("ok") is True, r2)

        print(f"    … 电脑关机了。等 {TTL + 2:.0f} 秒模拟“隔夜”（真实是 6 小时）")
        time.sleep(TTL + 2)

        h = health()
        check("④ 中转把超时设备清掉了（设计中如此）",
              dev not in [d["deviceId"] for d in h["devices"]])

        # 磁盘上还留着码绑定吗？
        sf = data / "relay_state.json"
        disk_codes = json.loads(sf.read_text(encoding="utf-8")).get("codes", [])
        disk_has = any(c.get("deviceId") == dev for c in disk_codes)
        print(f"    磁盘 relay_state.json 里还有这台设备的码绑定: {disk_has}")
        print(f"    内存里 codeCount={h.get('codeCount')}")

        r3 = reg(dev, code)
        check("⑤ 电脑第二天开机 —— 应该能连上", r3.get("ok") is True, r3)
        if not r3.get("ok"):
            print()
            print("    🔴 复现成功。返回: %s" % r3)

        print()
        print("═══ 追问：重启中转能不能救回来？ ═══")
        relay.terminate()
        relay.wait(timeout=15)
        relay = start()

        h2 = health()
        after = json.loads(sf.read_text(encoding="utf-8")).get("codes", [])
        after_has = any(c.get("deviceId") == dev for c in after)
        print(f"    重启后磁盘里这台设备的码绑定: {after_has}")
        print(f"    重启后内存 codeCount={h2.get('codeCount')}")
        print(f"    重启后 allowNew={h2.get('allowNew')}")

        r4 = reg(dev, code)
        check("⑥ 重启中转后电脑能连上（说明重启能救回来）",
              r4.get("ok") is True, r4)
        if not r4.get("ok"):
            print("    🔴 重启也救不回来 —— 管理员必须手动打开开关，"
                  "而这台设备在管理页上**已经看不到**（被清理了），"
                  "管理员根本不知道该放开谁。")

        print()
        print("═══ 追问：有没有绕过口子？ ═══")
        # 换个新 deviceId（模拟重装）
        r5 = reg("fresh-reinstall-999", "135790")
        check("⑦ 换新 deviceId 重装 —— 同样被挡（符合开关语义）",
              r5.get("ok") is False and r5.get("blocked") is True, r5)

        # 换用原来的码，但新 deviceId
        r6 = reg("fresh-reinstall-888", code)
        check("⑧ 新 deviceId + 老配对码 —— 也被挡",
              r6.get("ok") is False, r6)

        print()
        print("═══ 对照：打开开关后是否一切恢复 ═══")
        admin({"allow": True})
        r7 = reg(dev, code)
        check("⑨ 打开开关后原设备恢复", r7.get("ok") is True, r7)

        print()
        print("=" * 62)
        print(f"  通过 {passed} 项，失败 {failed} 项")
        print("=" * 62)
        return 0
    finally:
        if relay is not None and relay.poll() is None:
            relay.terminate()
            try:
                relay.wait(timeout=10)
            except Exception:                              # noqa: BLE001
                relay.kill()


if __name__ == "__main__":
    sys.exit(main())