ykxmz/_tests/test_verify_relay_xss.py

# -*- coding: utf-8 -*-
"""对抗性验证第二轮：修正上一版的归因错误，并加打真正的危险面。

上一版的错：场景 4 里五种非法 deviceId 用了**同一个配对码**，
第一个（控制字符）成功注册后，后面几种其实是被"码已被占用"拒掉的，
并不是因为 deviceId 非法 —— 归因错了，等于没测到。

这一版：
  * 每次尝试用**全新的配对码**，保证拒绝原因只可能来自 deviceId 本身
  * 加打 **管理页 XSS**：deviceId 和 name 是设备自己报上来的，
    如果管理页不转义就直接拼进 HTML，就是存储型 XSS
  * 加打 **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 = 8138
BASE = f"http://127.0.0.1:{PORT}"
_UA = {"User-Agent": "LanRemote/1.8.2"}
ADMIN_KEY = "test-admin-key-67890"

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="advrelay2_"))
    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": "3600",
            "RELAY_ADMIN_KEY": ADMIN_KEY,
            "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

        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:
                raw = exc.read().decode("utf-8", "replace")
                try:
                    return json.loads(raw)
                except Exception:                          # noqa: BLE001
                    return {"ok": False, "raw": raw[:160]}

        def admin_get(path: str) -> tuple:
            r = urllib.request.Request(
                BASE + path, headers={**_UA, "X-Admin": ADMIN_KEY})
            try:
                with urllib.request.urlopen(r, timeout=15) as resp:
                    return resp.status, resp.read().decode("utf-8", "replace")
            except urllib.error.HTTPError as exc:
                return exc.code, exc.read().decode("utf-8", "replace")

        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"))

        # 配对码生成器：保证每次都是全新的，避免"码被占"污染归因
        counter = [int(time.time()) % 100000]

        def fresh_code() -> str:
            counter[0] = (counter[0] + 1) % 1000000
            return "%06d" % counter[0]

        # ============================================================
        print()
        print("═══ 场景 A：非法 deviceId —— 每次用全新配对码，归因干净 ═══")
        # ============================================================
        cases = [
            ("超长 200 字符", "x" * 200, False),
            ("超长 81 字符", "y" * 81, False),
            ("正好 80 字符", "z" * 80, True),
            ("控制字符 NUL", "dev\x00abc", None),
            ("控制字符 \\r\\n", "dev\r\nInjected", None),
            ("HTML 标签", "<script>alert(1)</script>", None),
            ("引号", "dev\"'`", None),
            ("路径分隔符", "../../etc/passwd", None),
            ("Unicode", "设备-😀-ünïcode", None),
            ("只有空格", "   ", False),
            ("空字符串", "", False),
        ]
        for label, bad, expect in cases:
            r = reg(bad, fresh_code())
            got = r.get("ok") is True
            if expect is None:
                # 未定义期望：只记录实际行为，不判对错
                verdict = "接受" if got else "拒绝"
                print(f"    · {label:18s} -> {verdict}"
                      + ("" if got else f"（{r.get('error','')}）"))
                if got and ("<script" in bad or "\x00" in bad or "\r" in bad):
                    note(f"🟡 deviceId 接受危险字符：{label} 被接受"
                         f"（{bad!r}）。它会被拼进管理页 HTML，"
                         f"见场景 B 的 XSS 检查。")
            else:
                check(f"deviceId {label} {'应接受' if expect else '应拒绝'}",
                      got == expect, r)

        # ============================================================
        print()
        print("═══ 场景 B：管理页 XSS（deviceId / name 是设备自己报的）═══")
        # ============================================================
        xss_dev = '<img src=x onerror=alert("XSS-DEV")>'
        xss_name = '<script>alert("XSS-NAME")</script>'
        r = reg(xss_dev, fresh_code(), name=xss_name)
        print(f"    注册结果: ok={r.get('ok')}  error={r.get('error','')}")

        if r.get("ok"):
            st, page = admin_get("/zhongzhuanzhan")
            check("管理页能取到", st == 200, st)

            # 原始（可执行的）标签绝不能出现
            raw_img = '<img src=x onerror='
            raw_script = '<script>alert('
            check("管理页没有原始 <img onerror=...> 标签（未转义的 deviceId）",
                  raw_img not in page,
                  "发现原始标签 -> 存储型 XSS")
            check("管理页没有原始 <script>alert( 标签（未转义的 name）",
                  raw_script not in page,
                  "发现原始标签 -> 存储型 XSS")

            # 转义后的形式应该出现（说明确实渲染了这个设备）
            if "&lt;img" not in page and "&lt;script" not in page:
                note("🟡 管理页里既没有转义后的形式也没找到这个设备 —— "
                     "可能设备没被渲染，需要人工确认")
            else:
                print("    · 页面里出现的是转义形式（&lt;img / &lt;script），说明有转义")

            # 更细：onerror= 出现了但没有被转义上下文包裹
            if "onerror=" in page:
                idx = page.find("onerror=")
                ctx = page[max(0, idx - 60):idx + 60]
                print(f"    · onerror= 的上下文: {ctx!r}")

        # ============================================================
        print()
        print("═══ 场景 C：两台设备用同一个 deviceId ═══")
        # ============================================================
        dup = f"dup-{int(time.time()) % 100000}"
        c1, c2 = fresh_code(), fresh_code()
        ra = reg(dup, c1, name="第一台")
        rb = reg(dup, c2, name="第二台")
        print(f"    第一次: ok={ra.get('ok')} code={ra.get('code')}")
        print(f"    第二次: ok={rb.get('ok')} code={rb.get('code')} "
              f"note={rb.get('note','')}")
        h = health()
        rows = [d for d in h["devices"] if d["deviceId"] == dup]
        check("同一 deviceId 只产生一条设备记录", len(rows) == 1,
              f"{len(rows)} 条")
        if len(rows) == 1:
            check("该设备最终只占一个码（第二次的码没变成孤儿）",
                  h.get("codeCount") == len(h["devices"]),
                  f"codeCount={h.get('codeCount')} 设备数={len(h['devices'])}")
            # 第二次报的码 c2 应该被释放（如果被拒）或设备改用 c2
            used = rows[0]["code"]
            print(f"    · 设备最终用的码: {used}（c1={c1} c2={c2}）")
            # 拿没被用的那个码去注册新设备，验证它不是孤儿
            other = c1 if used == c2 else c2
            r = reg(f"probe-{int(time.time()) % 100000}", other)
            check(f"没被采用的码 {other} 没有变成孤儿",
                  r.get("ok") is True, r)

        # ============================================================
        print()
        print("═══ 场景 D：改码的边界 ═══")
        # ============================================================
        d1 = f"edge-{int(time.time()) % 100000}"
        own = fresh_code()
        reg(d1, own)

        others = fresh_code()
        reg(f"other-{int(time.time()) % 100000}", others)

        # D1: 改成别人占着的码
        body = json.dumps({"deviceId": d1, "code": others}).encode()
        rq = urllib.request.Request(
            BASE + "/v1/admin/set_code", data=body, method="POST",
            headers={**_UA, "X-Admin": ADMIN_KEY,
                     "Content-Type": "application/json"})
        try:
            with urllib.request.urlopen(rq, timeout=15) as resp:
                res = json.loads(resp.read().decode("utf-8"))
                st = resp.status
        except urllib.error.HTTPError as exc:
            st = exc.code
            try:
                res = json.loads(exc.read().decode("utf-8"))
            except Exception:                              # noqa: BLE001
                res = {}
        check("改成别的设备占用的码 -> 被拒",
              res.get("ok") is False or st >= 400, (st, res))

        # D2: 改成一个没人用的码，然后确认旧码释放
        newc = fresh_code()
        body = json.dumps({"deviceId": d1, "code": newc}).encode()
        rq = urllib.request.Request(
            BASE + "/v1/admin/set_code", data=body, method="POST",
            headers={**_UA, "X-Admin": ADMIN_KEY,
                     "Content-Type": "application/json"})
        with urllib.request.urlopen(rq, timeout=15) as resp:
            res2 = json.loads(resp.read().decode("utf-8"))
        check("改成空闲码 -> 成功", res2.get("ok") is True, res2)
        r = reg(f"reuser-{int(time.time()) % 100000}", own)
        check("改码后旧码可以被新设备占用（没留孤儿）",
              r.get("ok") is True, r)

        # D3: 给不存在的设备改码
        body = json.dumps({"deviceId": "no-such-device-xyz",
                           "code": fresh_code()}).encode()
        rq = urllib.request.Request(
            BASE + "/v1/admin/set_code", data=body, method="POST",
            headers={**_UA, "X-Admin": ADMIN_KEY,
                     "Content-Type": "application/json"})
        try:
            with urllib.request.urlopen(rq, timeout=15) as resp:
                res3 = json.loads(resp.read().decode("utf-8"))
                st3 = resp.status
        except urllib.error.HTTPError as exc:
            st3 = exc.code
            try:
                res3 = json.loads(exc.read().decode("utf-8"))
            except Exception:                              # noqa: BLE001
                res3 = {}
        # 注意：这里返回 400 是**正确**的（"没有这台设备"是客户端问题，
        # 不是服务端错误）。我第一版断言写成 st3 == 200 是我期望值错了 ——
        # 关键是要有明确错误信息、且不能是 5xx。
        check("给不存在的设备改码 -> 明确失败而不是 500",
              res3.get("ok") is False and 400 <= st3 < 500
              and bool(res3.get("error")), (st3, res3))

        # D4: 非 6 位码
        for bad_code in ("12345", "1234567", "abcdef", "", "12 456"):
            body = json.dumps({"deviceId": d1, "code": bad_code}).encode()
            rq = urllib.request.Request(
                BASE + "/v1/admin/set_code", data=body, method="POST",
                headers={**_UA, "X-Admin": ADMIN_KEY,
                         "Content-Type": "application/json"})
            try:
                with urllib.request.urlopen(rq, timeout=15) as resp:
                    res4 = json.loads(resp.read().decode("utf-8"))
                    st4 = resp.status
            except urllib.error.HTTPError as exc:
                st4 = exc.code
                try:
                    res4 = json.loads(exc.read().decode("utf-8"))
                except Exception:                          # noqa: BLE001
                    res4 = {}
            check(f"非法新码 {bad_code!r} -> 被拒且不是 500",
                  res4.get("ok") is False and st4 < 500, (st4, res4))

        print()
        print("=" * 62)
        print(f"  通过 {passed} 项，失败 {failed} 项")
        if findings:
            print()
            print("  发现的问题：")
            for f in findings:
                print("    - " + f)
        print("=" * 62)
        Path("_build/_adv_relay_result2.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())