mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 12:30:44 +00:00
xmz7891
b62ac865ea
被控端 - 修复 MTU 问题：夹 MSS + 路径 MTU 探测（公网可用率 48% -> 100%） - 复用 HTTP 连接（免掉每次新建 TLS 的 30% 握手失败） - 每分钟回头找局域网，自动切回快 400 倍的路径 - 画面心跳帧：静止时省流量，客户端仍知道流是活的 - 锁屏时给出可操作的提示 手表端 - 内置 GTS Root R4 / GlobalSign Root CA，修 SSLHandshakeException - 连接错误翻译成人话（证书/时间/解析/超时分别提示） - 经中转连上后用服务器报的地址刷新已配对记录 - 新增局域网连接页面、服务器延迟检测 服务器端 - 支持 chunked 请求体（修每 16 秒一次的连接错乱与大面积 502） - 网络调优脚本 net-tune.sh（开机自动应用） - 管理页：批量删除、网络体检、下载区 - 隧道改 HTTP/2 + 双连接 文档 - 许可协议 / 安装前必读 / 安装后须知（作者：小麦子）
287 lines
11 KiB
Python
287 lines
11 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("二、管理页上有删除按钮吗")
|
|
status, html = get("/", headers=admin)
|
|
check("管理页能打开", status == 200, status)
|
|
# 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))
|
|
|
|
status, again = post("/v1/agent/register", {
|
|
"deviceId": "pc-new", "code": "778899", "name": "新设备",
|
|
"version": "1.4.0"})
|
|
check("孤儿绑定确实挡着这个码（复现了用户的问题）",
|
|
again.get("ok") is False, 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)
|
|
check("报告清掉了孤儿绑定",
|
|
result.get("orphansCleared", 0) >= 1,
|
|
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())
|