ykxmz/_tests/test_relay_admin_login.py

# -*- coding: utf-8 -*-
"""验证管理页的登录流程。

流程：GET / -> 登录页（无数据）-> POST /admin/login 密码 -> 拿 Cookie
      -> 带 Cookie 访问 / 和 /v1/admin/devices 应该能进
      -> 密码错 -> 回登录页且带错误提示
"""
from __future__ import annotations

import http.cookiejar
import json
import sys
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path


# Cloudflare 的 Browser Integrity Check 会拦 Python 默认 UA
# （返回 403 + error code 1010），所以请求都要带上这个。
_UA = {"User-Agent": "LanRemote/1.4.0"}

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "_build"))

info = json.loads((ROOT / "_build" / "relay_info.json").read_text(
    encoding="utf-8"))
URL = info["url"]
# ⚠ PASSWORD 不能在这里赋值 —— _admin_password() 定义在下面，
#   模块级直接调用会 NameError（pyflakes 也会报 undefined name）。
#   所以在 helper 定义之后才取（见文件下方）。

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
# 拿不到就跳过需要登录的用例，而不是编一个假密码硬试。
# ---------------------------------------------------------------------------
# ---- 测试专用管理密码 ----------------------------------------------------
# 测试要登录它自己启动的中转，但密码是那边随机的 —— 两边对不上就会
# 全部 401（表现成"功能坏了"，实际是环境没打通）。
#
# 这里生成一个**本次运行专用**的随机密码，写进环境变量；
# 测试启动中转用的是 dict(os.environ)，所以会自动传给中转。
#
# 为什么不用写死的：写死就又变成"密码进仓库"了，白改一场。
import os as _os_for_pw
import secrets as _secrets_for_pw
TEST_ADMIN_PW = "t" + _secrets_for_pw.token_hex(12)
_os_for_pw.environ["RELAY_ADMIN_PASSWORD"] = TEST_ADMIN_PW
_os_for_pw.environ.setdefault("RELAY_ADMIN_USER", "xmz7891")
# ---------------------------------------------------------------------------


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
    # 本机路径都读不到（Windows 上访问不了设备的 /root），
    # 就 SSH 过去读 —— 密码在那个文件里，不在任何仓库文件里。
    try:
        import paramiko
        for host in ("192.168.2.156", "192.168.10.105"):
            try:
                c = paramiko.SSHClient()
                c.set_missing_host_key_policy(paramiko.AutoAddPolicy())
                c.connect(host, username="root", password="1313144",
                          timeout=8, look_for_keys=False, allow_agent=False)
                _in, out, _err = c.exec_command(
                    "cat /root/lanremote-relay/data/admin_password.txt",
                    timeout=15)
                pw = out.read().decode("utf-8", "replace").strip()
                c.close()
                if pw:
                    return pw
            except Exception:                             # noqa: BLE001
                continue
    except Exception:                                     # noqa: BLE001
        pass
    return ""


# helper 定义完了，这里才能取值
PASSWORD = _admin_password()



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


def main() -> int:
    print(f"中转地址: {URL}")
    print()

    jar = http.cookiejar.CookieJar()
    opener = urllib.request.build_opener(
        urllib.request.HTTPCookieProcessor(jar))

    print("一、没登录时")
    try:
        _req = urllib.request.Request(URL + "/", headers=_UA)
        with opener.open(_req, timeout=30) as resp:
            body = resp.read().decode("utf-8", "replace")
            status = resp.status
    except urllib.error.HTTPError as exc:
        status = exc.code
        body = exc.read().decode("utf-8", "replace")
    check("管理页返回 200（给登录页而不是错误）", status == 200, status)
    check("看到的是登录页", "管理密码" in body or "登录" in body,
          body[:80])
    check("登录前不泄露设备信息", "配对码" not in body, "泄露了")

    # 管理接口没登录
    try:
        _req = urllib.request.Request(URL + "/v1/admin/devices", headers=_UA)
        with opener.open(_req, timeout=20) as resp:
            status2 = resp.status
    except urllib.error.HTTPError as exc:
        status2 = exc.code
    check("管理接口没登录被拒", status2 == 401, status2)

    print()
    print("二、密码错")
    data = urllib.parse.urlencode({"username": "xmz7891", "password": "000000"}).encode()
    req = urllib.request.Request(URL + "/admin/login", data=data, method="POST", headers=_UA)
    req.add_header("Content-Type", "application/x-www-form-urlencoded")
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            bad_status = resp.status
            bad_body = resp.read().decode("utf-8", "replace")
    except urllib.error.HTTPError as exc:
        bad_status = exc.code
        bad_body = exc.read().decode("utf-8", "replace")
    check("错密码被拒（重定向回登录页）", bad_status in (200, 302, 303),
          bad_status)
    check("错密码后没种 Cookie", len(jar) == 0, f"{len(jar)} 个 cookie")

    print()
    print("三、密码对")
    data = urllib.parse.urlencode({"username": "xmz7891", "password": PASSWORD}).encode()
    req = urllib.request.Request(URL + "/admin/login", data=data, method="POST", headers=_UA)
    req.add_header("Content-Type", "application/x-www-form-urlencoded")
    with opener.open(req, timeout=30) as resp:
        login_status = resp.status
        resp.read()
    check("登录返回重定向", login_status in (200, 302), login_status)
    cookies = [c.name for c in jar]
    check("种下了 lr_admin Cookie", "lr_admin" in cookies, str(cookies))

    print()
    print("四、登录后")
    with opener.open(urllib.request.Request(URL + "/", headers=_UA), timeout=30) as resp:
        body = resp.read().decode("utf-8", "replace")
        status = resp.status
    check("能打开管理页", status == 200, status)
    check("看到的是后台而不是登录页",
          "配对码" in body or "登记设备" in body, body[:80])
    check("页面上有退出登录", "退出" in body or "logout" in body.lower(), "")

    with opener.open(urllib.request.Request(URL + "/v1/admin/devices", headers=_UA), timeout=30) as resp:
        snap = json.loads(resp.read())
    check("管理接口能拿到设备列表", snap.get("ok") is True, str(snap)[:80])

    print()
    print("五、退出登录")
    try:
        with opener.open(urllib.request.Request(URL + "/admin/logout", headers=_UA), timeout=30) as resp:
            resp.read()
    except urllib.error.HTTPError:
        pass
    try:
        with opener.open(urllib.request.Request(URL + "/v1/admin/devices", headers=_UA), timeout=20) as resp:
            after = resp.status
    except urllib.error.HTTPError as exc:
        after = exc.code
    check("退出后管理接口又被拒", after == 401, after)

    print()
    print("六、脚本通道（X-Admin 头仍可用）")
    req = urllib.request.Request(URL + "/v1/admin/devices", headers=_UA)
    req.add_header("X-Admin", PASSWORD)
    try:
        with urllib.request.urlopen(req, timeout=20) as resp:
            ok = resp.status
    except urllib.error.HTTPError as exc:
        ok = exc.code
    check("X-Admin 传密码也能用（方便脚本）", ok == 200, ok)

    print()
    print("=" * 60)
    print(f"通过 {passed} 项，失败 {failed} 项")
    return 0 if failed == 0 else 1


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