mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 10:10:43 +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 + 双连接 文档 - 许可协议 / 安装前必读 / 安装后须知（作者：小麦子）
255 lines
9.1 KiB
Python
255 lines
9.1 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""验证「手表能不能判断出电脑和自己同网络」的数据链路。
|
||
|
||
用户要求：
|
||
手表连上之后，如果发现电脑和自己**在同一个网络**，就切局域网直连；
|
||
不在同一个网络就继续用中转。
|
||
|
||
要判断这个，手表需要知道电脑的局域网 IP 和端口。链路是：
|
||
被控端上报 lanIps/port
|
||
-> 中转存在 Device 上
|
||
-> 快照和设备信息里带给手表
|
||
-> 手表拿自己的 IP 比网段
|
||
|
||
这个测试验证前三步（Java 那步靠编译 + 代码审查）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import shutil
|
||
import socket
|
||
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
|
||
sys.path.insert(0, str(ROOT / "windows"))
|
||
|
||
RELAY_PORT = 8096
|
||
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[:150]}
|
||
|
||
|
||
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 my_lan_ip() -> str:
|
||
"""本机在局域网的地址（和被测控端用的是同一招）。"""
|
||
try:
|
||
probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
try:
|
||
probe.connect(("8.8.8.8", 80))
|
||
return probe.getsockname()[0]
|
||
finally:
|
||
probe.close()
|
||
except OSError:
|
||
return ""
|
||
|
||
|
||
def main() -> int:
|
||
work = Path(tempfile.mkdtemp(prefix="lan_detect_"))
|
||
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
|
||
|
||
mine = my_lan_ip()
|
||
print(f" （本机局域网地址：{mine or '拿不到'}）")
|
||
|
||
print()
|
||
print("一、被控端上报局域网 IP 和端口")
|
||
status, first = post("/v1/agent/register", {
|
||
"deviceId": "pc-lan", "code": "445566", "name": "带局域网的电脑",
|
||
"version": "1.4.0",
|
||
"lanIps": ["192.168.2.100", "10.0.0.5"],
|
||
"port": 8765})
|
||
check("登记成功", first.get("ok") is True, first)
|
||
|
||
status, body = get("/v1/health")
|
||
devices = json.loads(body).get("devices", [])
|
||
target = next((d for d in devices if d.get("deviceId") == "pc-lan"),
|
||
None)
|
||
check("设备在列表里", target is not None, devices)
|
||
if target:
|
||
check("lanIps 存下来了",
|
||
target.get("lanIps") == ["192.168.2.100", "10.0.0.5"],
|
||
target.get("lanIps"))
|
||
check("port 存下来了", target.get("port") == 8765,
|
||
target.get("port"))
|
||
|
||
print()
|
||
print("二、手表接入时能拿到这两个值")
|
||
status, attach = post("/v1/client/attach",
|
||
{"code": "445566", "name": "手表"})
|
||
check("接入成功", attach.get("ok") is True, attach)
|
||
check("接入响应里有 lanIps",
|
||
attach.get("lanIps") == ["192.168.2.100", "10.0.0.5"],
|
||
attach.get("lanIps"))
|
||
check("接入响应里有 port", attach.get("port") == 8765,
|
||
attach.get("port"))
|
||
|
||
print()
|
||
print("三、脏数据要挡住（不能把乱七八糟的东西转给手表）")
|
||
post("/v1/agent/register", {
|
||
"deviceId": "pc-dirty", "code": "447788", "name": "脏数据",
|
||
"version": "1.4.0",
|
||
"lanIps": ["192.168.1.1", "not-an-ip", "999.1.1.1",
|
||
"<script>", "", "10.0.0.1", 12345],
|
||
"port": 99999})
|
||
status, body = get("/v1/health")
|
||
dirty = next((d for d in json.loads(body).get("devices", [])
|
||
if d.get("deviceId") == "pc-dirty"), None)
|
||
check("脏数据设备在列表里", dirty is not None)
|
||
if dirty:
|
||
check("只留下合法 IPv4",
|
||
dirty.get("lanIps") == ["192.168.1.1", "10.0.0.1"],
|
||
dirty.get("lanIps"))
|
||
check("非法端口归零（手表就不会试着直连）",
|
||
dirty.get("port") == 0, dirty.get("port"))
|
||
|
||
print()
|
||
print("四、被控端真的会算自己的局域网 IP")
|
||
from lanremote.config import Store
|
||
from lanremote.relay import RelayLink
|
||
store = Store(work / "store")
|
||
store.set_pairing_code("556677")
|
||
store.set("server_name", "本机")
|
||
store.set("port", 8877)
|
||
link = RelayLink(store, log=lambda _m: None)
|
||
ips = link._lan_ips()
|
||
check("算出了至少一个局域网 IP", len(ips) >= 1, ips)
|
||
check("算出来的都是 IPv4",
|
||
all(len(x.split(".")) == 4 for x in ips), ips)
|
||
check("不含回环地址",
|
||
all(not x.startswith("127.") for x in ips), ips)
|
||
if mine:
|
||
check("算出来的和本机实际地址一致（或包含它）",
|
||
mine in ips or any(
|
||
x.rsplit(".", 1)[0] == mine.rsplit(".", 1)[0]
|
||
for x in ips),
|
||
f"实际 {mine}，算出 {ips}")
|
||
|
||
print()
|
||
print("五、同网段判断逻辑（手表的算法，这里用 Python 复刻验证）")
|
||
def same_subnet(a, b):
|
||
pa, pb = a.split("."), b.split(".")
|
||
if len(pa) != 4 or len(pb) != 4:
|
||
return False
|
||
return pa[:3] == pb[:3]
|
||
|
||
cases = [
|
||
("192.168.2.10", "192.168.2.100", True),
|
||
("192.168.2.10", "192.168.3.100", False),
|
||
("10.0.0.5", "10.0.0.9", True),
|
||
("10.0.0.5", "192.168.1.9", False),
|
||
]
|
||
for a, b, expect in cases:
|
||
got = same_subnet(a, b)
|
||
check(f"{a} vs {b} -> {'同网段' if expect else '不同网段'}",
|
||
got is expect, got)
|
||
|
||
print()
|
||
print("六、挑地址：多个网卡时挑同网段的那个")
|
||
def pick(device_ips, my_ip):
|
||
for device_ip in device_ips:
|
||
if same_subnet(device_ip, my_ip):
|
||
return device_ip
|
||
return None
|
||
|
||
check("有两个候选时挑中同网段的",
|
||
pick(["10.0.0.5", "192.168.2.50"], "192.168.2.10")
|
||
== "192.168.2.50")
|
||
check("都不在同一个网段时返回空（继续用中转）",
|
||
pick(["10.0.0.5", "172.16.1.1"], "192.168.2.10") is None)
|
||
|
||
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())
|