mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 21:50: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 + 双连接 文档 - 许可协议 / 安装前必读 / 安装后须知（作者：小麦子）
225 lines
9.3 KiB
Python
225 lines
9.3 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""验证 Bug 1 的修复：安装时先查配对码，被占了就不往下走。
|
||
|
||
要验证：
|
||
1. 中转有只读查询接口 /v1/code/check，而且**不会占用那个码**
|
||
2. 已占用的码返回 taken=true，并带上占用的设备名
|
||
3. 空闲的码返回 taken=false
|
||
4. 被控端的 check_code_available() 能正确报告问题
|
||
5. run_setup 的流程是"先检查，通过才开自启/修防火墙"
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
from pathlib import Path
|
||
|
||
_UA = {"User-Agent": "LanRemote/1.4.0"}
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
sys.path.insert(0, str(ROOT / "windows"))
|
||
|
||
PORT = 8101
|
||
BASE = f"http://127.0.0.1:{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 get_json(path):
|
||
req = urllib.request.Request(BASE + path, headers=_UA)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=10) 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 register(device_id, code, name):
|
||
body = json.dumps({"deviceId": device_id, "code": code, "name": name,
|
||
"version": "1.4.0"}).encode()
|
||
req = urllib.request.Request(BASE + "/v1/agent/register", data=body,
|
||
method="POST",
|
||
headers={**_UA,
|
||
"Content-Type": "application/json"})
|
||
return json.loads(urllib.request.urlopen(req, timeout=10).read().decode())
|
||
|
||
|
||
def main() -> int:
|
||
work = Path(tempfile.mkdtemp(prefix="code_check_"))
|
||
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:
|
||
if get_json("/v1/health")[0] == 200:
|
||
up = True
|
||
break
|
||
except Exception: # noqa: BLE001
|
||
continue
|
||
check("本地中转起来了", up)
|
||
if not up:
|
||
return 1
|
||
|
||
print()
|
||
print("一、空闲的码查出来是空闲")
|
||
status, result = get_json("/v1/code/check?code=654321")
|
||
check("接口返回 200", status == 200, status)
|
||
check("ok=true", result.get("ok") is True, result)
|
||
check("taken=false", result.get("taken") is False, result)
|
||
|
||
print()
|
||
print("二、查询本身不能占用那个码")
|
||
register("pc-other", "654321", "别的电脑")
|
||
status, result = get_json("/v1/code/check?code=135790")
|
||
check("查一个没人用的码之后它还是空闲",
|
||
result.get("taken") is False, result)
|
||
# 再登记一次，确认 135790 没被查询动作占掉
|
||
reply = register("pc-probe", "135790", "验证机")
|
||
check("那个码仍然可以被正常登记", reply.get("ok") is True, reply)
|
||
|
||
print()
|
||
print("三、被占的码查出来是[已被占用]，还带设备名")
|
||
status, result = get_json("/v1/code/check?code=654321")
|
||
check("taken=true", result.get("taken") is True, result)
|
||
check("带上了占用的设备名",
|
||
result.get("deviceName") == "别的电脑", result.get("deviceName"))
|
||
|
||
print()
|
||
print("四、非法输入要挡住")
|
||
status, result = get_json("/v1/code/check?code=abc")
|
||
check("非数字返回 400", status == 400, status)
|
||
status, result = get_json("/v1/code/check?code=12345")
|
||
check("位数不够返回 400", status == 400, status)
|
||
status, result = get_json("/v1/code/check")
|
||
check("不给码返回 400", status == 400, status)
|
||
|
||
print()
|
||
print("五、被控端的 check_code_available()")
|
||
from lanremote.config import Store
|
||
# run_agent 是独立脚本（不在 lanremote 包里），按文件路径加载
|
||
import importlib.util
|
||
spec = importlib.util.spec_from_file_location(
|
||
"run_agent_mod", str(ROOT / "windows" / "run_agent.py"))
|
||
run_agent_mod = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(run_agent_mod)
|
||
check_code_available = run_agent_mod.check_code_available
|
||
|
||
store = Store(work / "store")
|
||
store.set("relay_url", BASE)
|
||
|
||
problem = check_code_available(store, "654321")
|
||
check("被占的码会返回一段说明", bool(problem) and
|
||
"已经被" in str(problem), problem)
|
||
check("说明里带设备名", "别的电脑" in str(problem), problem)
|
||
check("说明里告诉用户换一个", "换一个" in str(problem), problem)
|
||
|
||
problem = check_code_available(store, "999888")
|
||
check("空闲的码返回 None（放行）", problem is None, problem)
|
||
|
||
problem = check_code_available(store, "abc")
|
||
check("格式不对会报错", bool(problem), problem)
|
||
|
||
store.set("relay_url", "http://127.0.0.1:1") # 连不上的地址
|
||
problem = check_code_available(store, "654321")
|
||
check("中转连不上时放行（不能因为服务器挂了就装不了）",
|
||
problem is None, problem)
|
||
|
||
# 全新安装时 relay_url 是**空的**，要走内置默认值那条分支。
|
||
# 之前漏测了这条 —— 结果那个分支里用了没导入的 DEFAULT_RELAY_URL，
|
||
# 安装到最后一步直接 NameError 弹窗崩溃。
|
||
store.set("relay_url", "")
|
||
try:
|
||
problem = check_code_available(store, "654321")
|
||
check("中转地址为空时不崩溃（走内置默认值）", True, problem)
|
||
except NameError as exc:
|
||
check("中转地址为空时不崩溃（走内置默认值）", False,
|
||
f"NameError: {exc}")
|
||
except Exception as exc: # noqa: BLE001
|
||
check("中转地址为空时不崩溃（走内置默认值）", False,
|
||
f"{type(exc).__name__}: {exc}")
|
||
|
||
problem = check_code_available(store, "654321")
|
||
check("空地址时也返回值（不抛异常）", problem is None, problem)
|
||
|
||
print()
|
||
print("六、run_setup 的流程顺序对不对（源码级）")
|
||
src = (ROOT / "windows" / "run_agent.py").read_text(encoding="utf-8")
|
||
# 只在 run_setup 这个函数体里比顺序 ——
|
||
# 全局搜会搜到函数**定义**（定义在文件更早的位置），
|
||
# 那样比出来的顺序没有意义（一开始就这么写错过）。
|
||
body_start = src.find("def run_setup(store)")
|
||
body = src[body_start:]
|
||
idx_check = body.find("check_code_available(store, code)")
|
||
idx_fw = body.find("fix_firewall_with_flash(store)")
|
||
idx_auto = body.find('if result["autostart"]:')
|
||
check("run_setup 里先做配对码检查", idx_check > 0, idx_check)
|
||
check("检查在修防火墙之前（只看 run_setup 内部）",
|
||
0 < idx_check < idx_fw, f"检查@{idx_check} 防火墙@{idx_fw}")
|
||
check("检查也在开自启之前", 0 < idx_check < idx_auto,
|
||
f"检查@{idx_check} 自启@{idx_auto}")
|
||
# 被占时循环回去重开窗口：分支里只有 _tell_code_problem，
|
||
# 没有 break/return —— 也就是说它不会往下走到防火墙那一步。
|
||
# 被占之后的那一小段里（到 while 循环结束为止）
|
||
# 不能有 break/return —— 有的话就说明它继续往下走改防火墙了。
|
||
# 注意 break 在 if problem is None 里是**通过**分支，不算。
|
||
tell = body.find("_tell_code_problem(problem)")
|
||
after = body[tell:tell + 200] if tell > 0 else ""
|
||
check("被占时会告诉用户", tell > 0, after[:120])
|
||
check("告诉用户之后不往下走（那一段没有 break/return）",
|
||
"break" not in after and "return" not in after,
|
||
after[:160])
|
||
|
||
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())
|