DCU_DesktopController/scripts/test_lan_probe.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""test_lan_probe.py — 设备发现的离线自测（不需要真板子，可 CI 跑）。

覆盖三件事，缺一不可：
  1) 判据本身有灵敏度：同一套代码必须把"板子的 lite 响应"和"路由器的 HTML/
     缺字段 JSON"分开，也要把"旧固件（无 gd32_features）"收下而不是当反证扔掉。
  2) 真 socket 路径走真代码：假板服务绑在**本机 WLAN 网卡地址的高端口**上
     （不是 127.0.0.1 —— 用回环测不到 selectors/截止/refused 的真实语义）。
  3) 不自证：起/关假板服务，同一请求的判决必须翻转；种子故意指错地址，
     真在那儿的假板仍要被找到。

用法: python scripts/test_lan_probe.py        退出码 0=PASS 1=FAIL
"""
import io
import json
import os
import re
import socket
import sys
import threading
import urllib.request
from http.server import BaseHTTPRequestHandler, HTTPServer

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, ROOT)

import lan_probe as lp                      # noqa: E402

FAILS = []
PASSES = []


def check(name, cond, detail=""):
    if cond:
        PASSES.append(name)
        print("  PASS  " + name)
    else:
        FAILS.append(name)
        print("  FAIL  " + name + ("  :: " + str(detail) if detail else ""))


def lite_body(mode="STA", ip="192.168.31.127", connected=True, with_feat=True):
    b = {"wifi": {"mode": mode, "ip": ip, "connected": connected}}
    if with_feat:
        b["gd32_features"] = {"known": True, "flags": 15, "can_ch0": True,
                              "can_ch1": True, "can_ch2": True, "can_ch_mask": 7}
    return json.dumps(b)


# ───────────────────────── 1) 判决表（含反向对照） ─────────────────────────
def test_classify():
    print("[1] classify 判决表")
    ip = "192.168.31.127"
    r = lp.classify(ip, 200, "application/json", lite_body(ip=ip))
    check("正对照：真 lite 响应 → board/full",
          r["verdict"] == "board" and r["schema"] == "full" and r["kind"] == "sta", r)

    # 反向对照 1：路由器登录页
    r = lp.classify("192.168.31.1", 200, "text/html; charset=utf-8",
                    "<!DOCTYPE html><title>miwifi</title>")
    check("反对照1：HTML → rejected/not-json",
          r["verdict"] == "rejected" and r["reason"] == "not-json", r)

    # 反向对照 2：形状不对的 JSON
    r = lp.classify("192.168.31.9", 200, "application/json", '{"error":"unauthorized"}')
    check("反对照2：无 wifi 对象 → rejected/no-wifi-object",
          r["verdict"] == "rejected" and r["reason"] == "no-wifi-object", r)

    # 反向对照 3：STA 自称在别处（端口转发/捕获门户）
    r = lp.classify(ip, 200, "application/json", lite_body(ip="10.0.0.5"))
    check("反对照3：STA 自报 IP 不符 → rejected/ip-mismatch",
          r["verdict"] == "rejected" and r["reason"].startswith("ip-mismatch"), r)

    # 反向对照 4：HTTP 不是 200
    r = lp.classify(ip, 404, "text/plain", "not found")
    check("反对照4：404 → rejected/http-404",
          r["verdict"] == "rejected" and r["reason"] == "http-404", r)

    # 旧固件：有 wifi 无 gd32_features 必须**收下**（标 core-only），不能当反证丢
    r = lp.classify(ip, 200, "application/json", lite_body(ip=ip, with_feat=False))
    check("旧固件无 gd32_features → board/core-only（不丢弃）",
          r["verdict"] == "board" and r["schema"] == "core-only", r)

    # 关键坑：lite 在 NOT(STA&&connected) 时一律报 AP，所以 AP+connected:false
    # 多半是"STA 掉租约的板子"，既不能当在线 AP 板，也不能当反证 → unknown
    r = lp.classify(ip, 200, "application/json",
                    lite_body(mode="AP", ip="0.0.0.0", connected=False))
    check("AP+connected:false → unknown/ap-not-connected",
          r["verdict"] == "unknown" and r["reason"] == "ap-not-connected", r)

    # 真 AP 回退（PC 连了板子热点）：connected=true 才认
    r = lp.classify("192.168.31.1", 200, "application/json",
                    lite_body(mode="AP", ip="0.0.0.0", connected=True))
    check("AP+connected:true → board/kind=ap",
          r["verdict"] == "board" and r["kind"] == "ap", r)

    # busy/degraded 桩：完整 status 抢不到 mutex 时只回 version+ota.sys_state
    r = lp.classify(ip, 200, "application/json",
                    '{"version":"0.5.0","ota":{"sys_state":"busy"},"cpu":{}}')
    check("busy 桩（无 wifi） → rejected/no-wifi-object（ lite 不受此影响）",
          r["verdict"] == "rejected", r)

    # 兜底不许有未定义分支：mode 值不认识 → unknown
    r = lp.classify(ip, 200, "application/json", '{"wifi":{"mode":"NA","ip":null}}')
    check("mode 非法值 → unknown（兜底，不落 rejected）",
          r["verdict"] == "unknown", r)


# ───────────────────────── 2) IP 校验 ─────────────────────────
def test_ip_validation():
    print("[2] IPv4 校验（候选表与端点写入都靠它，判据在服务侧）")
    for s in ["192.168.31.127", "10.0.0.1", "1.2.3.4"]:
        check("收合法地址 " + s, lp.valid_ipv4(s))
    for s in ["not-an-ip", "1.2.3.4.5", "256.1.1.1", "", "192.168.31", "1.2.3.04x"]:
        check("拒 " + repr(s), not lp.valid_ipv4(s))


# ───────────────────────── 3) 真 socket + 真路由 ─────────────────────────
class _Fake(BaseHTTPRequestHandler):
    BODY = ""
    CTYPE = "application/json"
    CODE = 200

    def do_GET(self):
        b = self.BODY.encode("utf-8")
        self.send_response(self.CODE)
        self.send_header("Content-Type", self.CTYPE)
        self.send_header("Content-Length", str(len(b)))
        self.end_headers()
        self.wfile.write(b)

    def log_message(self, *a):
        pass


def _serve(ip, body, ctype="application/json", code=200):
    cls = type("F", (_Fake,), {"BODY": body, "CTYPE": ctype, "CODE": code})
    srv = HTTPServer((ip, 0), cls)
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    return srv, srv.server_address[1]


def local_ip():
    n = lp.local_net()
    return n.get("default_src_ip"), n


def _free_port(host_ip):
    """要一个"确实没人监听"的端口：绑 0 拿到号→关掉→再 connect 确认被拒。
    别拿 bp+1 猜：Windows 会把 ephemeral 端口连着发，猜中的就是隔壁那个服务。"""
    for _ in range(20):
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.bind((host_ip, 0))
        p = s.getsockname()[1]
        s.close()
        t = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        t.settimeout(0.3)
        try:
            refused = t.connect_ex((host_ip, p)) != 0
        finally:
            t.close()
        if refused:
            return p
    return None


def test_probe_route():
    print("[3] 真 socket：/api/lan/probe 走真代码路径")
    ip, net = local_ip()
    check("本机默认路由取到 WLAN 地址（期望 192.168.31.x）",
          bool(ip) and ip.startswith("192.168.31."), net)
    if not ip:
        return

    board_body = lite_body(ip=ip)          # 自报 IP 必须等于被探地址
    board, bp = _serve(ip, board_body)
    router, rp = _serve(ip, "<!DOCTYPE html><p>miwifi</p>", "text/html")
    closed = _free_port(ip)                 # 实测被拒的端口
    check("拿到一个确实没人监听的端口", closed is not None, closed)
    if closed is None:
        return

    httpd, _p = __import__("esp32_server")._bind_server(0)
    port = httpd.server_address[1]
    threading.Thread(target=httpd.serve_forever, daemon=True).start()

    def call(body, timeout=20):
        req = urllib.request.Request(
            "http://127.0.0.1:%d/api/lan/probe" % port,
            data=json.dumps(body).encode("utf-8"),
            headers={"Content-Type": "application/json"})
        return json.loads(urllib.request.urlopen(req, timeout=timeout).read().decode("utf-8"))

    r = call({"ips": [ip], "ports": [bp, rp, closed], "local": True, "timeout_ms": 900})
    by = {(x["ip"], x["port"]): x for x in r["results"]}
    check("假板（真 socket，绑本机网卡）→ board",
          by[(ip, bp)]["verdict"] == "board", by[(ip, bp)])
    check("假路由器 → rejected/not-json",
          by[(ip, rp)]["verdict"] == "rejected" and by[(ip, rp)]["reason"] == "not-json",
          by[(ip, rp)])
    check("没人监听的端口 → unknown（不是 rejected：没有反证，只是没答）",
          by[(ip, closed)]["verdict"] == "unknown", by[(ip, closed)])
    check("tcpOpen 不把 refused 算成开着",
          r["scan"]["tcpOpen"] == 2, r["scan"])
    check("local 段随响应返回", bool(r.get("local", {}).get("cidr")), r.get("local"))

    # 灵敏度对照：关掉假板，同一请求的判决必须翻
    board.shutdown()
    r2 = call({"ips": [ip], "ports": [bp], "timeout_ms": 600})
    check("灵敏度对照：假板下电后 board → unknown",
          r2["results"][0]["verdict"] != "board", r2["results"][0])

    # 配置自证对照：种子故意指一个错地址，真在那儿的假路由器地址仍要如实分类
    r3 = call({"ips": [ip], "ports": [rp], "timeout_ms": 600})
    check("种子指错也不会把非板子说成板子", r3["results"][0]["verdict"] == "rejected",
          r3["results"][0])

    # 非法 IP / 路径白名单
    r4 = call({"ips": ["bogus", "1.2.3.4.5"], "ports": [80]})
    check("非法 IP 被剔且带原因", r4["scan"]["candidates"] == 0
          and len(r4["scan"]["skipped"]) == 2, r4["scan"])
    r5 = call({"ips": [ip], "ports": [rp], "path": "/logs"})
    check("危险路径不进白名单（/logs 会排空固件日志环）",
          "/system/status?lite=1" == r5["path"] and any(
              "path-not-allowed" in e for e in r5["errors"]), r5)

    # 连续 10 批不泄漏状态：每批结果数必须齐（3 个相同 IP 去重 × 2 个端口 = 2 条）
    ok, detail = True, ""
    for _ in range(10):
        rr = call({"ips": [ip] * 3, "ports": [rp, closed], "timeout_ms": 400})
        if len(rr["results"]) != 2:
            ok = False
            detail = "期望 2 条，实际 %d 条" % len(rr["results"])
            break
    check("连续多批结果数稳定（批间无残留状态、去重按 ip:port）", ok, detail)

    httpd.shutdown()
    router.shutdown()


# ───────────────────────── 4) 结构回归 ─────────────────────────
def test_structure():
    print("[4] 结构回归（把已经踩过的坑钉住）")
    src = io.open(os.path.join(ROOT, "lan_probe.py"), encoding="utf-8").read()
    check("lan_probe.py 不写 stdout/stderr（--window 下子进程 stderr 是父进程管道）",
          "print(" not in src and "sys.stderr" not in src)
    m = re.search(r"^ALLOWED_PATHS = \((.*?)\)", src, re.M)
    allowed = m.group(1) if m else ""
    check("探测白名单只有安全路径（/logs 会排空固件日志环，/system/probe 会发真实 SPI 事务）",
          m is not None and "/logs" not in allowed and "/system/probe" not in allowed
          and "PROBE_PATH" in allowed and '"/system/status"' in allowed, allowed)

    srv = io.open(os.path.join(ROOT, "esp32_server.py"), encoding="utf-8").read()
    check("路由挂在 do_POST（GET 会被无扩展名当 HTML 的分支吃掉）",
          srv.count('"/api/lan/probe"') >= 1
          and "_handle_lan_probe" in srv)
    check("QuietServer.handle_error 仍静默（异常灌进父进程 4KB 管道会冻住服务）",
          re.search(r"def handle_error\(self[^)]*\):\s*\n\s+pass", srv) is not None)

    js = {}
    for rel in ["js/common/base-url.js", "js/common/device-link.js", "js/can.js",
                "js/can/dbc.js", "js/common/lan-picker.js", "js/app.js"]:
        js[rel] = io.open(os.path.join(ROOT, rel), encoding="utf-8").read()

    writers = [k for k, v in js.items() if "saveDeviceIp(" in v.replace("function saveDeviceIp(", "")]
    check("写端点的只有 device-link（+ base-url 自身实现）",
          set(writers) <= {"js/common/base-url.js", "js/common/device-link.js"}, writers)
    check("can.js 不再把设备自报 IP 回写成端点（0.0.0.0 污染源）",
          "setDeviceIp" not in js["js/can.js"] and "setDeviceIp" not in js["js/can/dbc.js"])
    check("can.js 改为只提示不符", "与当前端点" in js["js/can.js"])
    check("device-link 里端点唯一写手 + 刷写门",
          "selectDevice: function" in js["js/common/device-link.js"]
          and "flashBusyNow()" in js["js/common/device-link.js"])
    check("自动补扫门齐（failStreak + 刷写 + ECAT 环跑 + 冷却）",
          "maybeAutoScan" in js["js/common/device-link.js"]
          and "snap.links.ecat.loop" in js["js/common/device-link.js"])
    check("面板不重算判据（只取服务侧 verdict 字段，不自己解析 gd32_features）",
          re.search(r"gd32_features\s*[\[.]", js["js/common/lan-picker.js"]) is None
          and ".verdict" in js["js/common/lan-picker.js"])
    check("当前端点不进候选表（不重复占设备 httpd 的 socket）",
          "当前端点" in js["js/common/lan-picker.js"])


def main():
    print("== lan_probe 自测 ==")
    test_classify()
    test_ip_validation()
    test_probe_route()
    test_structure()
    print("\n合计 PASS=%d FAIL=%d" % (len(PASSES), len(FAILS)))
    if FAILS:
        print("FAILED: " + "; ".join(FAILS))
        return 1
    print("ALL PASS")
    return 0


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