ykxmz/_tests/test_verify_relay_edges.py

# -*- coding: utf-8 -*-
"""对抗性验证：中转状态机的边界场景（verifier 独立编写）。

不依赖任何人的自述，直接起一个真中转、打真接口、看真结果。

重点打这些：
    1. **DEVICE_TTL 清理 + allow_new=False 的死亡组合**
       设备离线超过 DEVICE_TTL 后被 snapshot 清掉，
       而 register() 判断"是不是老设备"用的是 self.devices ——
       那台设备再回来会不会被当成新设备、被开关永久挡死？

    2. **改码之后 codes / devices / state.json 三者自洽**
       同一台设备连续改码 A->B->C，中间态会不会留孤儿？

    3. **snapshot 的清理顺序**
       会不会出现「codes 清了但 devices 还在」或反过来？

    4. **deviceId 塞奇怪字符**
       超长、控制字符、路径分隔符、Unicode。

    5. **并发改码**
       两个线程同时给同一台设备改成不同的码，会不会出现两个码都指向它。

    6. **未鉴权访问管理接口**
       不带 cookie / 不带 X-Admin 能不能改状态。

用 RELAY_DEVICE_TTL 把超时压到几秒，就能在测试里真实触发清理路径，
不用等 6 小时。
"""
from __future__ import annotations

import http.cookiejar
import json
import os
import subprocess
import sys
import tempfile
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
PORT = 8137
BASE = f"http://127.0.0.1:{PORT}"
TTL = 3.0                       # 设备 3 秒不上报就算超时
_UA = {"User-Agent": "LanRemote/1.8.2"}

passed = 0
failed = 0
findings: list[str] = []


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 note(text: str) -> None:
    findings.append(text)
    print("  [!!] " + text)


def main() -> int:
    work = Path(tempfile.mkdtemp(prefix="advrelay_"))
    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),
            "RELAY_DEVICE_TTL": str(TTL),
            "RELAY_ADMIN_KEY": "test-admin-key-12345",
            "RELAY_POLL": "2",
            "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(60):
            time.sleep(0.5)
            try:
                r = urllib.request.Request(BASE + "/v1/health", headers=_UA)
                with urllib.request.urlopen(r, timeout=3) as resp:
                    if resp.status == 200:
                        up = True
                        break
            except Exception:                              # noqa: BLE001
                continue
        check("真中转起来了", up)
        if not up:
            return 1

        ADMIN = {"**": "**"}

        def admin_post(path: str, payload: dict) -> tuple:
            body = json.dumps(payload).encode()
            r = urllib.request.Request(BASE + path, data=body, method="POST",
                                       headers={**_UA, "X-Admin":
                                                "test-admin-key-12345",
                                                "Content-Type":
                                                    "application/json"})
            try:
                with urllib.request.urlopen(r, timeout=20) as resp:
                    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[:160]}

        def health() -> dict:
            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: str, code: str, name: str = "对抗测试机") -> dict:
            body = json.dumps({"deviceId": dev, "code": code, "name": name,
                               "version": "1.8.2",
                               "lanIps": ["192.168.1.5"],
                               "port": 8765}).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[:160]}

        stamp = int(time.time()) % 100000

        # ============================================================
        print()
        print("═══ 场景 1：DEVICE_TTL 清理 + allow_new=False 的死亡组合 ═══")
        # ============================================================
        dev1 = f"ttl-dev-{stamp}"
        code1 = "%06d" % (stamp % 1000000)
        r1 = reg(dev1, code1)
        check("设备首次注册成功", r1.get("ok") is True, r1)

        st, res = admin_post("/v1/admin/allow_new", {"allow": False})
        check("关掉「允许新设备接入」成功",
              st == 200 and res.get("allowNew") is False, (st, res))

        # 立刻再注册一次（设备还在 devices 里）—— 应该放行
        r2 = reg(dev1, code1)
        check("【对照】设备还在 devices 里时，老设备能通过开关",
              r2.get("ok") is True, r2)

        # 等超过 TTL，然后调 health 触发 snapshot 清理
        print(f"    等 {TTL + 2:.0f} 秒让设备超时…")
        time.sleep(TTL + 2)
        h = health()                       # snapshot() 在这里跑清理
        devs_now = [d["deviceId"] for d in h.get("devices", [])]
        check("【前提】超时设备已被 snapshot 清出 devices",
              dev1 not in devs_now, devs_now)
        check("【前提】该设备的码绑定也已被清（codes 成对清理）",
              h.get("codeCount") == 0, h.get("codeCount"))

        # 关键一击：设备"回来"了
        r3 = reg(dev1, code1)
        if r3.get("ok") is True:
            check("超时清理后老设备仍能重新上线", True)
        else:
            check("超时清理后老设备仍能重新上线", False, r3)
            note("🔴 **严重**：设备离线超过 DEVICE_TTL 被清理后，"
                 "若 allow_new=False，该设备再回来会被当成新设备**永久拒绝**。"
                 f" 复现：register({dev1!r}) -> allow_new=False -> 等 {TTL+2:.0f}s "
                 f"-> GET /v1/health -> register 同 deviceId 得到 {r3}")

        # 打开开关再试，确认确实是开关挡的
        admin_post("/v1/admin/allow_new", {"allow": True})
        r4 = reg(dev1, code1)
        check("【归因】打开开关后同一 deviceId 立刻能注册（证明是被开关挡的）",
              r4.get("ok") is True, r4)

        # ============================================================
        print()
        print("═══ 场景 2：改码 A->B->C 后 codes 里有没有留孤儿 ═══")
        # ============================================================
        dev2 = f"code-dev-{stamp}"
        codeA = "%06d" % ((stamp + 111) % 1000000)
        codeB = "%06d" % ((stamp + 222) % 1000000)
        codeC = "%06d" % ((stamp + 333) % 1000000)
        ra = reg(dev2, codeA)
        check("设备注册（码 A）", ra.get("ok") is True, ra)

        st, res = admin_post("/v1/admin/set_code",
                             {"deviceId": dev2, "code": codeB})
        check("改码 A->B 成功", st == 200 and res.get("ok"), (st, res))
        st, res = admin_post("/v1/admin/set_code",
                             {"deviceId": dev2, "code": codeC})
        check("改码 B->C 成功", st == 200 and res.get("ok"), (st, res))

        h = health()
        cnt = h.get("codeCount")
        dev2_codes = [d["code"] for d in h["devices"]
                      if d["deviceId"] == dev2]
        check("一台设备只占一个码（codeCount 与设备数一致）",
              cnt == len(h["devices"]), f"codeCount={cnt} 设备数={len(h['devices'])}")
        check("设备当前码是 C", dev2_codes == [codeC], dev2_codes)

        # A 和 B 应该都能被别的设备抢走（如果还是孤儿就抢不走）
        dev3 = f"code-dev3-{stamp}"
        rA = reg(dev3, codeA)
        check("旧码 A 已被释放（另一台设备能占用）", rA.get("ok") is True, rA)

        # ============================================================
        print()
        print("═══ 场景 3：改码之后重启，codes 是否还原成自洽状态 ═══")
        # ============================================================
        st, res = admin_post("/v1/admin/set_code",
                             {"deviceId": dev2, "code": codeB})
        check("再改成 B", st == 200 and res.get("ok"), (st, res))
        state_file = data / "relay_state.json"
        raw = json.loads(state_file.read_text(encoding="utf-8"))
        bound = [c for c in raw.get("codes", [])
                 if c.get("deviceId") == dev2]
        check("state.json 里该设备只有一条码绑定",
              len(bound) == 1, bound)

        # ============================================================
        print()
        print("═══ 场景 4：deviceId 塞奇怪字符 ═══")
        # ============================================================
        weird = [
            ("超长 200 字符", "x" * 200),
            ("控制字符", "dev\x00\x01\x02"),
            ("路径分隔符", "../../etc/passwd"),
            ("Unicode", "设备-😀-ünïcode"),
            ("只有空格", "   "),
        ]
        for label, bad in weird:
            try:
                r = reg(bad, "%06d" % ((stamp + 444) % 1000000))
                ok_rejected = r.get("ok") is False
                check(f"非法 deviceId（{label}）被拒绝而不是崩溃",
                      ok_rejected, r)
            except Exception as exc:                       # noqa: BLE001
                check(f"非法 deviceId（{label}）被拒绝而不是崩溃",
                      False, f"抛异常 {type(exc).__name__}: {exc}")

        # ============================================================
        print()
        print("═══ 场景 5：并发改码（两个线程同时改同一台设备）═══")
        # ============================================================
        dev5 = f"race-dev-{stamp}"
        reg(dev5, "%06d" % ((stamp + 555) % 1000000))
        codeX = "%06d" % ((stamp + 666) % 1000000)
        codeY = "%06d" % ((stamp + 777) % 1000000)
        errors: list = []

        def changer(target: str) -> None:
            try:
                admin_post("/v1/admin/set_code",
                           {"deviceId": dev5, "code": target})
            except Exception as exc:                       # noqa: BLE001
                errors.append(exc)

        t1 = threading.Thread(target=changer, args=(codeX,))
        t2 = threading.Thread(target=changer, args=(codeY,))
        t1.start()
        t2.start()
        t1.join()
        t2.join()
        check("并发改码没抛异常", not errors, errors)

        h = health()
        owned = [c for c, d in [("?", None)] ] if False else None
        dev5_row = [d for d in h["devices"] if d["deviceId"] == dev5]
        check("并发后设备还在", len(dev5_row) == 1, dev5_row)
        cur = dev5_row[0]["code"] if dev5_row else "?"
        check("并发后设备当前码是 X 或 Y 之一",
              cur in (codeX, codeY), cur)

        # 另一台设备能不能用"输掉"的那个码？能就说明没留孤儿
        dev6 = f"race-dev6-{stamp}"
        loser = codeY if cur == codeX else codeX
        r = reg(dev6, loser)
        check("并发改码后输掉的那个码没有变成孤儿（别的设备能占）",
              r.get("ok") is True, r)

        # ============================================================
        print()
        print("═══ 场景 6：管理接口未鉴权访问 ═══")
        # ============================================================
        for path, payload in (
            ("/v1/admin/allow_new", {"allow": False}),
            ("/v1/admin/set_code", {"deviceId": dev1, "code": "999999"}),
            ("/v1/admin/delete", {"deviceId": dev1}),
            ("/v1/admin/delete_all", {}),
        ):
            body = json.dumps(payload).encode()
            r = urllib.request.Request(BASE + path, data=body, method="POST",
                                       headers={**_UA, "Content-Type":
                                                "application/json"})
            try:
                with urllib.request.urlopen(r, timeout=10) as resp:
                    code = resp.status
                    resp.read()
            except urllib.error.HTTPError as exc:
                code = exc.code
                try:
                    exc.read()
                except Exception:                          # noqa: BLE001
                    pass
            except Exception as exc:                       # noqa: BLE001
                code = f"异常 {type(exc).__name__}"
            check(f"未鉴权 POST {path} 被拒（401/403）", code in (401, 403),
                  f"实际 {code}")

        # ============================================================
        print()
        print("═══ 场景 7：设备列表为空时管理页是否 500 ═══")
        # ============================================================
        admin_post("/v1/admin/delete_all", {})
        r = urllib.request.Request(BASE + "/zhongzhuanzhan",
                                   headers={**_UA, "X-Admin":
                                            "test-admin-key-12345"})
        try:
            with urllib.request.urlopen(r, timeout=10) as resp:
                page = resp.read().decode("utf-8", "replace")
                ok_page = resp.status == 200 and len(page) > 200
        except urllib.error.HTTPError as exc:
            page = exc.read().decode("utf-8", "replace")
            ok_page = False
            note(f"管理页在设备列表为空时返回 {exc.code}")
        check("设备列表为空时管理页仍正常渲染", ok_page)
        check("空列表页面没有 traceback",
              "Traceback" not in page and "NameError" not in page)

        # ============================================================
        print()
        print("═══ 场景 8：关掉开关后重启中转，开关状态是否还在 ═══")
        # ============================================================
        admin_post("/v1/admin/allow_new", {"allow": False})
        h1 = health()
        check("关掉开关", h1.get("allowNew") is False, h1.get("allowNew"))
        relay.terminate()
        relay.wait(timeout=15)
        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:
                h2 = health()
                up = True
                break
            except Exception:                              # noqa: BLE001
                continue
        check("重启后中转起来了", up)
        if up:
            check("重启后 allowNew 仍为 False（持久化生效）",
                  h2.get("allowNew") is False, h2.get("allowNew"))

        # ============================================================
        print()
        print("=" * 62)
        print(f"  通过 {passed} 项，失败 {failed} 项")
        if findings:
            print()
            print("  发现的问题：")
            for f in findings:
                print("    - " + f)
        print("=" * 62)
        Path("_build/_adv_relay_result.txt").write_text(
            f"passed={passed} failed={failed}\n"
            + "\n".join(findings) + "\n", encoding="utf-8")
        return 1 if failed else 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())