ykxmz/_tests/test_admin_password_gen.py

# -*- coding: utf-8 -*-
r'''验证「管理密码随机生成」这个改动。

要确认：
  1. 全新部署（没有 admin_password.txt）时，会生成一个**随机**密码，
     而不是某个写死的值
  2. 生成的密码写进了 data/admin_password.txt，权限 600
  3. 用那个密码能登录管理页
  4. 重启之后密码不变（从文件读回来）
  5. **同一个密码不会在两次全新部署里重复**（真的是随机的）
  6. 源码和成品里都搜不到老密码
'''
from __future__ import annotations

import http.cookiejar
import json
import os
import re
import shutil
import socket
import stat
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
UA = {"User-Agent": "LanRemote/1.8.2"}

passed = 0
failed = 0
OLD_PASSWORD = "REVA" + "edu<改从环境变量读>"      # 拼出来，免得这个文件自己又成一个泄漏点


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


def free_port() -> int:
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    p = s.getsockname()[1]
    s.close()
    return p


def start_relay(port: int, data: Path):
    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_POLL": "3",
                "PYTHONIOENCODING": "utf-8", "PYTHONUNBUFFERED": "1"})
    log = open(data.parent / ("relay-%d.log" % port), "wb")
    proc = subprocess.Popen(
        [sys.executable, "-u", str(ROOT / "relay" / "relay.py")], env=env,
        stdout=log, stderr=subprocess.STDOUT)
    base = "http://127.0.0.1:%d" % port
    for _ in range(80):
        time.sleep(0.4)
        try:
            with urllib.request.urlopen(
                    urllib.request.Request(base + "/v1/health", headers=UA),
                    timeout=4) as resp:
                if resp.status == 200:
                    return proc, base
        except Exception:                                 # noqa: BLE001
            continue
    return proc, base


def fresh_password(work: Path, tag: str) -> tuple:
    """起一台全新中转，返回 (密码, 数据目录, 进程, base)。"""
    port = free_port()
    data = work / tag
    proc, base = start_relay(port, data)
    time.sleep(1.0)
    pf = data / "admin_password.txt"
    pw = pf.read_text(encoding="utf-8").strip() if pf.is_file() else ""
    return pw, data, proc, base


def main() -> int:
    print("=" * 60)
    print("  管理密码随机生成 · 验证")
    print("=" * 60)

    work = Path(tempfile.mkdtemp(prefix="pwgen_"))
    procs = []
    try:
        # ---------------------------------------------------------- 一
        print("  ---- 一、全新部署会生成随机密码 ----")
        pw1, data1, p1, base1 = fresh_password(work, "fresh1")
        procs.append(p1)
        check("生成了密码文件", bool(pw1), "没有 admin_password.txt")
        check("密码不是那个老密码", pw1 != OLD_PASSWORD, "居然还是老密码！")
        check("密码长度 >= 12", len(pw1) >= 12, "只有 %d 位：%r" % (len(pw1), pw1))
        check("密码是纯字母数字（好念）", pw1.isalnum(), repr(pw1))

        pf = data1 / "admin_password.txt"
        if pf.is_file():
            mode = stat.S_IMODE(pf.stat().st_mode)
            if os.name == "nt":
                print("  [SKIP] 文件权限检查（Windows 的 chmod 语义不同）")
            else:
                check("密码文件权限是 600", mode == 0o600, oct(mode))

        # ---------------------------------------------------------- 二
        print("  ---- 二、密码能登录管理页 ----")
        jar = http.cookiejar.CookieJar()
        opener = urllib.request.build_opener(
            urllib.request.HTTPCookieProcessor(jar))
        form = urllib.parse.urlencode(
            {"username": "xmz7891", "password": pw1}).encode()
        req = urllib.request.Request(base1 + "/admin/login", data=form,
                                     method="POST", headers=UA)
        req.add_header("Content-Type", "application/x-www-form-urlencoded")
        try:
            with opener.open(req, timeout=20) as resp:
                st = resp.status
                resp.read()
        except urllib.error.HTTPError as exc:
            st = exc.code
        check("用生成的密码能登录", st in (200, 302), "状态 %s" % st)
        check("种下了 lr_admin Cookie",
              any(c.name == "lr_admin" for c in jar),
              [c.name for c in jar])

        # ---------------------------------------------------------- 三
        print("  ---- 三、老密码不能用了 ----")
        jar2 = http.cookiejar.CookieJar()
        opener2 = urllib.request.build_opener(
            urllib.request.HTTPCookieProcessor(jar2))
        form = urllib.parse.urlencode(
            {"username": "xmz7891", "password": OLD_PASSWORD}).encode()
        req = urllib.request.Request(base1 + "/admin/login", data=form,
                                     method="POST", headers=UA)
        req.add_header("Content-Type", "application/x-www-form-urlencoded")
        try:
            with opener2.open(req, timeout=20) as resp:
                st2 = resp.status
                resp.read()
        except urllib.error.HTTPError as exc:
            st2 = exc.code
        check("老密码登不进去", not any(c.name == "lr_admin" for c in jar2),
              "居然种下了 Cookie！状态 %s" % st2)

        # ---------------------------------------------------------- 四
        print("  ---- 四、重启后密码不变 ----")
        p1.terminate()
        try:
            p1.wait(timeout=8)
        except Exception:                                 # noqa: BLE001
            p1.kill()
        port1 = int(base1.rsplit(":", 1)[1])
        p1b, base1b = start_relay(port1, data1)
        procs.append(p1b)
        time.sleep(1.0)
        pw1b = (data1 / "admin_password.txt").read_text(
            encoding="utf-8").strip()
        check("重启后密码还是原来那个", pw1b == pw1, "%r != %r" % (pw1b, pw1))

        # ---------------------------------------------------------- 五
        print("  ---- 五、两次全新部署的密码不一样（真随机）----")
        pw2, data2, p2, _b2 = fresh_password(work, "fresh2")
        procs.append(p2)
        check("第二个部署也有密码文件", bool(pw2), "没有")
        check("两台全新部署的密码不同", pw1 != pw2,
              "两台都是 %r —— 说明不是随机的！" % pw1)

        # ---------------------------------------------------------- 六
        print("  ---- 六、源码里没有那个老密码 ----")
        hits = []
        for pat in ("relay/*.py", "windows/**/*.py", "_tests/*.py",
                    "_build/*.py", "installer/*"):
            for f in ROOT.glob(pat):
                if not f.is_file():
                    continue
                try:
                    text = f.read_text(encoding="utf-8", errors="replace")
                except Exception:                         # noqa: BLE001
                    continue
                if OLD_PASSWORD in text:
                    hits.append(str(f.relative_to(ROOT)))
        check("源码里搜不到老密码", not hits, "还有：%s" % hits[:6])
    finally:
        for p in procs:
            try:
                p.terminate()
                p.wait(timeout=5)
            except Exception:                             # noqa: BLE001
                try:
                    p.kill()
                except Exception:                         # noqa: BLE001
                    pass
        shutil.rmtree(work, ignore_errors=True)

    print()
    print("=" * 60)
    print("  通过 %d 项，失败 %d 项" % (passed, failed))
    return 0 if failed == 0 else 1


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