mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 17:10:43 +00:00
xmz7891
430923dd84
== 问题 ==
relay/relay.py 第 138 行：
ADMIN_PASSWORD_DEFAULT = "@REVAedu120830"
而且 _tests/ 和 _build/ 下 11 个文件里也写着明文密码。
这个仓库是要推到**公开的 Gitee / EzHub** 的，等于：
git 仓库 = 公开的
管理页路径 /zhongzhuanzhan = 公开的（就写在代码里）
公网地址 = 公开的
密码 = 也在仓库里
**= 把管理台的门钥匙挂在门上。**
而且 git 历史删不掉 —— 只是在最新提交里删掉没用，
读过仓库的人已经从历史里拿到了。
== 修法 ==
### 1. 代码里不再有任何固定密码
relay.py 的 admin_password() 改成和 admin_key 一样的做法：
环境变量 RELAY_ADMIN_PASSWORD
-> data/admin_password.txt（600 权限，不进仓库）
-> 都没有就**首次启动随机生成**，写进那个文件并打印一次
这样全新部署的密码天然是随机的、每台不同。
### 2. 11 个文件里的明文密码清掉
_tests/ acceptance.py, test_relay_admin_login.py,
test_admin_batch_delete.py, test_admin_button.py,
test_admin_device_control.py, test_admin_js.py,
test_admin_netcheck.py
_build/ check_web.sh, verify_site_admin.sh,
split_site_admin2.py, test_response_headers.py
统一改成从 RELAY_ADMIN_PASSWORD 环境变量读，
拿不到就从部署的 data/admin_password.txt 读，
再拿不到就**跳过需要登录的用例**（而不是编个假密码硬试）。
### 3. 真机上的密码已经轮换
设备上的 data/admin_password.txt 已换成用户指定的新密码，
并实测：
✓ 新密码能登录（HTTP 302 + lr_admin cookie）
✓ 老密码登不上了
✓ 新密码无换行（printf 不带 \n，免得读出来多一个字符）
顺带：重启中断了 relay-stability 的 A/B 对照 A 组（69 个样本），
分析数据时会标注这一段。
== 新增测试 ==
_tests/test_admin_password_gen.py 验证六件事：
1. 全新部署会生成随机密码（不是那个老密码）
2. 密码长度 >= 12、纯字母数字
3. 用生成的密码能登录管理页
4. **老密码登不进去**
5. 重启后密码不变（从文件读回来）
6. **两次全新部署的密码不一样**（真的是随机的）
== 还没做的（等用户决定）==
git **历史**里还有那个密码。要彻底清掉得重写历史 + 强制推送，
这是个危险操作，按用户的规矩必须他确认之后才做。
而且**光删历史不够** —— 密码已经公开过，必须换（已换）。
310 lines
13 KiB
Python
310 lines
13 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""验证中转管理器的「删除设备（连同配对码）」。
|
||
|
||
用户要求：
|
||
"这个中转管理器，首先删除设备，就是删除设备并删除对应的配对码"
|
||
|
||
要验证：
|
||
1. 删设备 -> 设备从列表消失
|
||
2. 删设备 -> 配对码绑定也清了（别的设备能用这个码了）
|
||
3. 删设备 -> 它的会话被踢掉
|
||
4. 按 code 删也行
|
||
5. delete_all 能一次清光
|
||
6. 管理页上有删除按钮
|
||
7. 删不存在的设备要报错，不能崩
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
from pathlib import Path
|
||
|
||
_UA = {"User-Agent": "LanRemote/1.4.0"}
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
RELAY_PORT = 8094
|
||
BASE = f"http://127.0.0.1:{RELAY_PORT}"
|
||
|
||
passed = 0
|
||
failed = 0
|
||
|
||
|
||
def check(name, ok, detail=""):
|
||
global passed, failed
|
||
if ok:
|
||
passed += 1
|
||
print(" [PASS] " + name)
|
||
else:
|
||
failed += 1
|
||
print(" [FAIL] " + name + " " + str(detail))
|
||
|
||
|
||
def post(path, payload=None, headers=None):
|
||
body = json.dumps(payload or {}, ensure_ascii=False).encode("utf-8")
|
||
hdrs = dict(_UA)
|
||
hdrs["Content-Type"] = "application/json"
|
||
hdrs.update(headers or {})
|
||
req = urllib.request.Request(BASE + path, data=body, method="POST",
|
||
headers=hdrs)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=15) as resp:
|
||
return resp.status, json.loads(resp.read().decode("utf-8"))
|
||
except urllib.error.HTTPError as exc:
|
||
raw = exc.read().decode("utf-8", "replace")
|
||
try:
|
||
return exc.code, json.loads(raw)
|
||
except Exception: # noqa: BLE001
|
||
return exc.code, {"raw": raw[:120]}
|
||
|
||
|
||
def get(path, headers=None):
|
||
hdrs = dict(_UA)
|
||
hdrs.update(headers or {})
|
||
req = urllib.request.Request(BASE + path, headers=hdrs)
|
||
try:
|
||
with urllib.request.urlopen(req, 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 main() -> int:
|
||
work = Path(tempfile.mkdtemp(prefix="admin_delete_"))
|
||
relay = None
|
||
try:
|
||
relay_data = work / "relaydata"
|
||
relay_data.mkdir(parents=True, exist_ok=True)
|
||
env = dict(os.environ)
|
||
env.update({
|
||
"RELAY_PORT": str(RELAY_PORT), "RELAY_BIND": "127.0.0.1",
|
||
"RELAY_DATA": str(relay_data), "RELAY_POLL": "6",
|
||
"PYTHONIOENCODING": "utf-8", "PYTHONUNBUFFERED": "1",
|
||
})
|
||
relay = subprocess.Popen(
|
||
[sys.executable, "-u", str(ROOT / "relay" / "relay.py")], env=env,
|
||
stdout=open(work / "relay.log", "wb"),
|
||
stderr=subprocess.STDOUT)
|
||
up = False
|
||
for _ in range(40):
|
||
time.sleep(0.5)
|
||
try:
|
||
status, _ = get("/v1/health")
|
||
if status == 200:
|
||
up = True
|
||
break
|
||
except Exception: # noqa: BLE001
|
||
continue
|
||
check("本地中转起来了", up)
|
||
if not up:
|
||
print((work / "relay.log").read_text(encoding="utf-8",
|
||
errors="replace")[-500:])
|
||
return 1
|
||
|
||
admin_key = (relay_data / "admin_key.txt").read_text().strip()
|
||
admin = {"X-Admin": admin_key}
|
||
|
||
print()
|
||
print("一、准备：登记两台设备")
|
||
post("/v1/agent/register", {"deviceId": "pc-A", "code": "111111",
|
||
"name": "电脑A", "version": "1.4.0"})
|
||
post("/v1/agent/register", {"deviceId": "pc-B", "code": "222222",
|
||
"name": "电脑B", "version": "1.4.0"})
|
||
status, body = get("/v1/health")
|
||
devices = json.loads(body).get("devices", [])
|
||
check("两台设备都登记上了", len(devices) == 2, len(devices))
|
||
|
||
print()
|
||
print("二、管理页上有删除按钮吗")
|
||
# ⚠ 管理页搬到 /zhongzhuanzhan 了（用户要求）。
|
||
# 这里原来写的是 get("/")，抓到的是**官网首页**，
|
||
# 于是「找不到删除按钮」——而这条测试还不在回归套件里，
|
||
# 所以一直没人发现它是红的。
|
||
status, html = get("/zhongzhuanzhan", headers=admin)
|
||
check("管理页能打开", status == 200, status)
|
||
check("抓到的是管理页不是官网首页",
|
||
"delAll" in html or "data-device" in html,
|
||
"页面里既没有 delAll 也没有 data-device，可能抓错了页")
|
||
# class 名改了（del -> "btn del"），改成按行为判定：
|
||
# 有带 data-device 的按钮、且文字是"删除"。
|
||
check("有「删除」按钮",
|
||
"data-device" in html and "删除" in html,
|
||
"找不到删除按钮")
|
||
check("有「删除全部设备」按钮", "delAll" in html, "没找到 delAll")
|
||
check("删除前的确认提示里说明了会清配对码",
|
||
"配对码绑定" in html or "清掉它的配对码" in html)
|
||
|
||
print()
|
||
print("三、按 deviceId 删一台")
|
||
status, result = post("/v1/admin/delete", {"deviceId": "pc-A"},
|
||
headers=admin)
|
||
check("删除返回 ok", result.get("ok") is True, result)
|
||
check("删的是 A", result.get("name") == "电脑A", result.get("name"))
|
||
check("它占了 1 条码绑定",
|
||
len(result.get("codesCleared") or []) == 1,
|
||
result.get("codesCleared"))
|
||
|
||
status, body = get("/v1/health")
|
||
devices = json.loads(body).get("devices", [])
|
||
check("A 从设备列表消失了", len(devices) == 1, len(devices))
|
||
check("剩下的是 B", devices and devices[0]["name"] == "电脑B",
|
||
devices)
|
||
|
||
print()
|
||
print("四、A 的配对码要能被别的设备用了")
|
||
status, again = post("/v1/agent/register", {
|
||
"deviceId": "pc-C", "code": "111111", "name": "电脑C",
|
||
"version": "1.4.0"})
|
||
check("新设备可以用刚释放的 111111",
|
||
again.get("ok") is True, again)
|
||
check("拿到的就是 111111", again.get("code") == "111111",
|
||
again.get("code"))
|
||
|
||
print()
|
||
print("五、按 code 删")
|
||
status, result = post("/v1/admin/delete", {"code": "222222"},
|
||
headers=admin)
|
||
check("按码删也能删掉", result.get("ok") is True, result)
|
||
check("删的是 B", result.get("name") == "电脑B", result.get("name"))
|
||
|
||
print()
|
||
print("六、删不存在的设备要报错，不能崩")
|
||
status, result = post("/v1/admin/delete", {"deviceId": "不存在"},
|
||
headers=admin)
|
||
check("返回 404", status == 404, status)
|
||
check("带了错误说明", bool(result.get("error")), result)
|
||
status, result = post("/v1/admin/delete", {}, headers=admin)
|
||
check("什么都不给返回 400", status == 400, status)
|
||
|
||
print()
|
||
print("七、delete_all 一次清光")
|
||
status, result = post("/v1/admin/delete_all", headers=admin)
|
||
check("delete_all 返回 ok", result.get("ok") is True, result)
|
||
status, body = get("/v1/health")
|
||
devices = json.loads(body).get("devices", [])
|
||
check("设备列表空了", len(devices) == 0, len(devices))
|
||
status, again = post("/v1/agent/register", {
|
||
"deviceId": "pc-D", "code": "111111", "name": "电脑D",
|
||
"version": "1.4.0"})
|
||
check("清光之后 111111 也可以用了",
|
||
again.get("ok") is True, again)
|
||
|
||
print()
|
||
print("八、删除后状态要落盘（重启不复活）")
|
||
post("/v1/agent/register", {"deviceId": "pc-E", "code": "333333",
|
||
"name": "电脑E", "version": "1.4.0"})
|
||
post("/v1/admin/delete", {"deviceId": "pc-E"}, headers=admin)
|
||
state_file = relay_data / "relay_state.json"
|
||
state = json.loads(state_file.read_text(encoding="utf-8"))
|
||
codes = [item.get("code") for item in state.get("codes", [])]
|
||
check("落盘的状态里没有 333333", "333333" not in codes, codes)
|
||
|
||
print()
|
||
print("九、孤儿绑定也要能清掉")
|
||
# 造一个"设备下线了但码还占着"的状态：
|
||
# 登记一台设备，然后直接把它从 devices 里摘掉（模拟超时清理），
|
||
# 但保留 codes 里的绑定。
|
||
post("/v1/agent/register", {"deviceId": "pc-orphan",
|
||
"code": "778899", "name": "孤儿",
|
||
"version": "1.4.0"})
|
||
state_file = relay_data / "relay_state.json"
|
||
before = json.loads(state_file.read_text(encoding="utf-8"))
|
||
codes_before = [i.get("code") for i in before.get("codes", [])]
|
||
check("778899 已经有绑定了", "778899" in codes_before, codes_before)
|
||
|
||
# 制造孤儿：把设备记录摘掉，只留绑定
|
||
import subprocess as _sp
|
||
_sp.run([sys.executable, "-c", f"""
|
||
import json
|
||
p = {str(state_file)!r}
|
||
d = json.load(open(p))
|
||
# 保留 778899，其余清掉，模拟"只剩绑定没有设备"
|
||
d["codes"] = [c for c in d.get("codes", []) if c.get("code") == "778899"]
|
||
json.dump(d, open(p, "w"))
|
||
"""], check=False)
|
||
# 重启中转，让它从盘上加载这个孤儿状态
|
||
relay.terminate()
|
||
relay.wait(timeout=8)
|
||
relay = subprocess.Popen(
|
||
[sys.executable, "-u", str(ROOT / "relay" / "relay.py")], env=env,
|
||
stdout=open(work / "relay2.log", "wb"),
|
||
stderr=subprocess.STDOUT)
|
||
for _ in range(30):
|
||
time.sleep(0.5)
|
||
try:
|
||
if get("/v1/health")[0] == 200:
|
||
break
|
||
except Exception: # noqa: BLE001
|
||
continue
|
||
|
||
status, body = get("/v1/health")
|
||
devices = json.loads(body).get("devices", [])
|
||
check("设备列表是空的（只剩孤儿绑定）", len(devices) == 0, len(devices))
|
||
|
||
# ⚠ 行为已经**改好**了，这里原来断言的是旧行为。
|
||
#
|
||
# 用户报过的问题：「设备下线了，配对码还占着，别的设备用不了」。
|
||
# 修法是在 snapshot() 里把「指向已不存在设备的码」当废码直接扔
|
||
# （见 relay.py 里 orphan 那一段），**而且顺手落盘** ——
|
||
# 不然内存清了、盘上还在，重启又恢复，每次重启回滚一次。
|
||
#
|
||
# 所以现在不需要管理员手动清场：上面那次 get("/v1/health")
|
||
# 已经触发了 snapshot()，孤儿码当场就没了。
|
||
state_now = json.loads(state_file.read_text(encoding="utf-8"))
|
||
codes_now = [i.get("code") for i in state_now.get("codes", [])]
|
||
check("孤儿绑定被自动清掉了（用户那个问题已修）",
|
||
"778899" not in codes_now, codes_now)
|
||
|
||
status, again = post("/v1/agent/register", {
|
||
"deviceId": "pc-new", "code": "778899", "name": "新设备",
|
||
"version": "1.4.0"})
|
||
check("孤儿清掉后这个码能直接用（不用管理员出手）",
|
||
again.get("ok") is True, again)
|
||
|
||
admin_key = (relay_data / "admin_key.txt").read_text().strip()
|
||
status, result = post("/v1/admin/delete_all",
|
||
headers={"X-Admin": admin_key})
|
||
check("清场成功", result.get("ok") is True, result)
|
||
# 孤儿已经在上面被自动清掉了，所以这里清的是刚注册的 pc-new
|
||
check("清场把绑定清光了",
|
||
len(json.loads(state_file.read_text(
|
||
encoding="utf-8")).get("codes", [])) == 0,
|
||
result.get("orphansCleared"))
|
||
|
||
status, body = get("/v1/health")
|
||
state = json.loads(state_file.read_text(encoding="utf-8"))
|
||
check("落盘里没有残留绑定了",
|
||
len(state.get("codes", [])) == 0, state.get("codes"))
|
||
|
||
status, again = post("/v1/agent/register", {
|
||
"deviceId": "pc-final", "code": "778899", "name": "终于能用了",
|
||
"version": "1.4.0"})
|
||
check("清场之后这个码能用了", again.get("ok") is True, again)
|
||
|
||
print()
|
||
print("=" * 62)
|
||
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__":
|
||
raise SystemExit(main())
|