mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 20:40:43 +00:00
xmz7891
9b17129c49
上一版把明文密码换成随机生成时，引入了三个问题。都是我自己的补丁
造成的，而且症状都像"功能坏了"，实际是环境和脚本的问题。
== bug 1：兜底替换把函数调用塞进了字符串 ==
fix_hardcoded_password.py 里有句兜底：
text = text.replace(OLD, '_admin_password()')
而原代码是：
{"username": "xmz7891", "password": "@<老密码>"}
^ @ 在引号**里面**
正则要求引号紧贴老密码，这里中间有个 @，所以没匹配上；
兜底替换就把老密码换成了函数调用，结果成了**字符串字面量**：
{"username": "xmz7891", "password": "@_admin_password()"}
7 个文件被这样弄坏。它们拿这串东西当密码去登录，
当然登不上 -> 401「请先登录」-> 管理接口测试全红（12 项失败）。
教训：**批量替换里的兜底 str.replace 太危险**，它不看上下文。
替换完必须检查有没有"函数调用被引号包起来"这种形状。
== bug 2：测试和它自己起的中转对不上密码 ==
密码改成随机生成之后，测试分成两边：
* 测试进程：_admin_password() 去环境变量/文件里找
* 它启动的中转：找不到就随机生成一个
两边不一致 -> 登录失败。
修法：测试在**模块级**（import 时）生成一个本次运行专用的随机密码，
写进 os.environ["RELAY_ADMIN_PASSWORD"]。测试启动中转用的是
dict(os.environ)，所以会自动传过去 —— 两边就是同一个密码。
用随机而不是写死：写死就又变成"密码进仓库"了，白改一场。
== bug 3：全新部署启动后没有密码文件 ==
admin_password() 原来只在有人访问登录页时才调用，于是全新部署
启动后 data/admin_password.txt 一直不存在，日志里也看不到密码，
管理员不知道该用什么登。
修法：main() 里在 admin_key() 后面补一句 admin_password()，
和钥匙一样的道理 —— 启动时生成并打印一次。
== 验证 ==
管理页设备控制 23/23 通过
管理页批删 24/24 通过
管理页按钮 10/10 通过
管理页 JS 通过
管理页网络体检 15/15 通过
密码随机生成 11/11 通过（含"老密码登不进去"、"两次部署密码不同"）
工作区搜不到明文密码了（git 历史里还有，等用户确认后重写）。
229 lines
8.8 KiB
Python
229 lines
8.8 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""验证管理页的"网络体检"接口 /v1/admin/netcheck。"""
|
||
from __future__ import annotations
|
||
|
||
import http.cookiejar
|
||
import json
|
||
import os
|
||
import re
|
||
import shutil
|
||
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.4.0"}
|
||
PORT = 8140
|
||
BASE = f"http://127.0.0.1:{PORT}"
|
||
|
||
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
|
||
return ""
|
||
|
||
|
||
|
||
def _has(html: str, token: str) -> bool:
|
||
"""判断 html 里有没有这个属性，**带不带引号都算**。
|
||
|
||
旧版页面写的是 class=pick（无引号），
|
||
重做后改成了标准的 class="pick"（带引号）。
|
||
测试不该因为引号差异就报失败 —— HTML 两种都合法。
|
||
"""
|
||
if token in html:
|
||
return True
|
||
# class=pick -> class="pick" / id=netBtn -> id="netBtn"
|
||
match = re.match(r"^([A-Za-z-]+)=(.+)$", token)
|
||
if not match:
|
||
return False
|
||
attr, value = match.group(1), match.group(2)
|
||
return ('%s="%s"' % (attr, value)) in html
|
||
|
||
|
||
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:
|
||
work = Path(tempfile.mkdtemp(prefix="netcheck_"))
|
||
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),
|
||
"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(40):
|
||
time.sleep(0.5)
|
||
try:
|
||
with urllib.request.urlopen(urllib.request.Request(
|
||
BASE + "/v1/health", headers=UA), timeout=2) as resp:
|
||
if resp.status == 200:
|
||
up = True
|
||
break
|
||
except Exception: # noqa: BLE001
|
||
continue
|
||
check("本地中转起来了", up)
|
||
if not up:
|
||
return 1
|
||
|
||
# 造两台设备，其中一台带 lanIps
|
||
for index, lan in ((1, ["192.168.10.103"]), (2, [])):
|
||
body = json.dumps({
|
||
"deviceId": f"pc-{index}", "code": f"90000{index}",
|
||
"name": f"电脑{index}", "version": "1.4.0",
|
||
"lanIps": lan, "port": 8765,
|
||
}).encode()
|
||
urllib.request.urlopen(urllib.request.Request(
|
||
BASE + "/v1/agent/register", data=body, method="POST",
|
||
headers={**UA, "Content-Type": "application/json"}),
|
||
timeout=10).read()
|
||
|
||
print()
|
||
print("一、没登录不能调")
|
||
try:
|
||
code = urllib.request.urlopen(urllib.request.Request(
|
||
BASE + "/v1/admin/netcheck", headers=UA), timeout=8).status
|
||
except urllib.error.HTTPError as exc:
|
||
code = exc.code
|
||
check("未登录被拒（401）", code == 401, code)
|
||
|
||
print()
|
||
print("二、登录后能拿到体检信息")
|
||
jar = http.cookiejar.CookieJar()
|
||
opener = urllib.request.build_opener(
|
||
urllib.request.HTTPCookieProcessor(jar))
|
||
form = urllib.parse.urlencode({"username": "xmz7891", "password": _admin_password()}).encode()
|
||
req = urllib.request.Request(BASE + "/admin/login", data=form,
|
||
method="POST", headers=UA)
|
||
req.add_header("Content-Type", "application/x-www-form-urlencoded")
|
||
try:
|
||
opener.open(req, timeout=15).read()
|
||
except urllib.error.HTTPError:
|
||
pass
|
||
|
||
with opener.open(urllib.request.Request(
|
||
BASE + "/v1/admin/netcheck", headers=UA), timeout=10) as resp:
|
||
info = json.loads(resp.read().decode("utf-8"))
|
||
print(f" 返回: {json.dumps(info, ensure_ascii=False)[:180]}")
|
||
|
||
check("ok=true", info.get("ok") is True, info)
|
||
check("有 deviceCount 信息（devices）", "devices" in info, info)
|
||
check("设备数正确（2）", info.get("devices") == 2, info.get("devices"))
|
||
check("有 tunnelMetrics 标记", "tunnelMetrics" in info, info)
|
||
check("有内存信息", "memoryKB" in info, info)
|
||
check("有公网地址字段", "publicUrl" in info, info)
|
||
check("统计了带局域网地址的设备",
|
||
info.get("lanDevices") == 1, info.get("lanDevices"))
|
||
# 本机一般没跑 cloudflared，所以这里应该是 False —— 关键是别崩
|
||
check("读不到隧道指标时不崩（tunnelMetrics=False）",
|
||
info.get("tunnelMetrics") is False, info.get("tunnelMetrics"))
|
||
# 关键：本地没有 cloudflared 时，读不到是正常的，
|
||
# 但**不能是因为代码写错**（比如少 import）。
|
||
# 之前就是漏了 import urllib.request，报 AttributeError，
|
||
# 而读不到时不崩这条断言照样通过 —— 断言得太松了。
|
||
err = str(info.get("tunnelError") or "")
|
||
check("读不到的原因不是代码错误（不能是 AttributeError/NameError）",
|
||
err not in ("AttributeError", "NameError", "TypeError"),
|
||
err)
|
||
|
||
print()
|
||
print("三、管理页上有这张卡片")
|
||
html = opener.open(urllib.request.Request(BASE + "/zhongzhuanzhan", headers=UA),
|
||
timeout=10).read().decode("utf-8")
|
||
check("有网络体检面板", _has(html, "id=netPanel"), "")
|
||
check("有「测一下」按钮", _has(html, "id=netBtn"), "")
|
||
check("JS 里调 netcheck 接口", "/v1/admin/netcheck" in html, "")
|
||
check("会显示隧道连接数", "tunnelConnections" in html, "")
|
||
|
||
print()
|
||
print("=" * 58)
|
||
print(f"通过 {passed} 项，失败 {failed} 项")
|
||
return 0 if failed == 0 else 1
|
||
finally:
|
||
if relay is not None:
|
||
try:
|
||
relay.terminate()
|
||
relay.wait(timeout=8)
|
||
except Exception: # noqa: BLE001
|
||
try:
|
||
relay.kill()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
time.sleep(0.5)
|
||
shutil.rmtree(work, ignore_errors=True)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|