mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 16:00: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 + 双连接 文档 - 许可协议 / 安装前必读 / 安装后须知（作者：小麦子）
368 lines
12 KiB
Python
368 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""网络自检 —— 确认手表的数据包到底有没有到达这台电脑。
|
||
|
||
这个工具回答一个二选一的问题：
|
||
|
||
手表的连接请求，究竟「到了电脑但被拒绝」，还是「压根没到」？
|
||
|
||
用法：
|
||
由 ykxmz.exe --netcheck 调用（已打包进 EXE，不需要装 Python）
|
||
|
||
会监听 90 秒，可以按 Ctrl+C 提前结束。
|
||
|
||
编码说明：
|
||
脚本启动时会把控制台代码页设成 UTF-8(65001) 再输出，
|
||
所以不论系统默认代码页是什么，中文都不会乱码。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import socket
|
||
import subprocess
|
||
import sys
|
||
import threading
|
||
import time
|
||
from datetime import datetime
|
||
|
||
TCP_PORT = 8765
|
||
UDP_PORT = 8766
|
||
LISTEN_SECONDS = 90
|
||
|
||
_NO_WINDOW = getattr(subprocess, "CREATE_NO_WINDOW", 0)
|
||
|
||
# WSAEADDRINUSE / WSAEACCES
|
||
_ADDR_IN_USE = (10048, 10013)
|
||
|
||
|
||
def setup_console() -> None:
|
||
"""把控制台和标准输出都设成 UTF-8，避免中文乱码。"""
|
||
try:
|
||
import ctypes
|
||
ctypes.windll.kernel32.SetConsoleOutputCP(65001)
|
||
ctypes.windll.kernel32.SetConsoleCP(65001)
|
||
except Exception:
|
||
pass
|
||
for stream in (sys.stdout, sys.stderr):
|
||
try:
|
||
stream.reconfigure(encoding="utf-8", errors="replace")
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def line(text: str = "") -> None:
|
||
print(text, flush=True)
|
||
|
||
|
||
def title(text: str) -> None:
|
||
line()
|
||
line("=" * 62)
|
||
line(f" {text}")
|
||
line("=" * 62)
|
||
|
||
|
||
def stamp() -> str:
|
||
return datetime.now().strftime("%H:%M:%S")
|
||
|
||
|
||
def local_ips() -> list[tuple[str, str]]:
|
||
found: list[tuple[str, str]] = []
|
||
try:
|
||
probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
probe.connect(("8.8.8.8", 80))
|
||
addr = probe.getsockname()[0]
|
||
probe.close()
|
||
if addr and not addr.startswith("127."):
|
||
found.append((addr, "默认路由"))
|
||
except OSError:
|
||
pass
|
||
try:
|
||
for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET):
|
||
addr = info[4][0]
|
||
if addr and not addr.startswith("127.") and addr not in [a for a, _ in found]:
|
||
found.append((addr, "主机名"))
|
||
except OSError:
|
||
pass
|
||
return found
|
||
|
||
|
||
def run(cmd: list[str], timeout: int = 15) -> str:
|
||
try:
|
||
proc = subprocess.run(cmd, capture_output=True, timeout=timeout,
|
||
creationflags=_NO_WINDOW,
|
||
encoding="mbcs", errors="replace")
|
||
return (proc.stdout or "") + (proc.stderr or "")
|
||
except Exception as exc: # noqa: BLE001
|
||
return f"({exc})"
|
||
|
||
|
||
def agent_running() -> bool:
|
||
"""被控端是不是正在运行（它会占着 8765/8766）。"""
|
||
text = run(["tasklist", "/FI", "IMAGENAME eq ykxmz.exe"])
|
||
return "ykxmz" in text
|
||
|
||
|
||
def stop_agent() -> bool:
|
||
"""结束被控端进程，好让自检能绑上端口。"""
|
||
run(["taskkill", "/IM", "ykxmz.exe", "/F"])
|
||
time.sleep(1.5)
|
||
return not agent_running()
|
||
|
||
|
||
def ask_stop_agent() -> bool:
|
||
"""询问是否关掉被控端。非交互环境直接返回 False。"""
|
||
if not sys.stdin.isatty():
|
||
return False
|
||
line()
|
||
line(" 被控端正在运行，占着 8765/8766，本自检绑不上端口。")
|
||
line(" 要抓包必须先让它让出端口。")
|
||
line()
|
||
try:
|
||
answer = input(" >>> 现在自动关掉被控端并继续抓包吗？(Y/N) ").strip().lower()
|
||
except Exception:
|
||
return False
|
||
if not answer.startswith("y"):
|
||
line(" 好，那本次不抓包。")
|
||
line(" （这不是故障 —— 被控端在跑就说明它能收到包；")
|
||
line(" 想抓包时先在它窗口点「退出」再重跑本脚本。）")
|
||
return False
|
||
if stop_agent():
|
||
line(" 已关闭被控端，继续抓包。")
|
||
line(" 抓完记得重新启动它。")
|
||
return True
|
||
line(" [警告] 关闭失败，可能是权限不够。请手动在它窗口点「退出」。")
|
||
return False
|
||
|
||
|
||
def show_network() -> str:
|
||
title("1. 本机网络")
|
||
ips = local_ips()
|
||
if ips:
|
||
for addr, tag in ips:
|
||
line(f" 本机 IP：{addr} [{tag}]")
|
||
line()
|
||
line(f" >>> 手表上就填这个 IP：{ips[0][0]} <<<")
|
||
else:
|
||
line(" 没能读到本机 IP（网络可能没连上）")
|
||
|
||
line()
|
||
wlan = run(["netsh", "wlan", "show", "interfaces"])
|
||
ssid = None
|
||
for raw in wlan.splitlines():
|
||
text = raw.strip()
|
||
if text.startswith("SSID") and "BSSID" not in text:
|
||
ssid = text.split(":", 1)[-1].strip()
|
||
if ssid:
|
||
line(f" 当前 WiFi：{ssid}")
|
||
line(" （手表必须连的是同一个 WiFi）")
|
||
else:
|
||
line(" 当前不是 WiFi 连接，或有线连接")
|
||
|
||
return ips[0][0] if ips else ""
|
||
|
||
|
||
def show_ports() -> None:
|
||
title("2. 端口占用情况")
|
||
tcp = run(["netstat", "-ano", "-p", "TCP"])
|
||
udp = run(["netstat", "-ano", "-p", "UDP"])
|
||
tcp_hit = [l.strip() for l in tcp.splitlines()
|
||
if f":{TCP_PORT}" in l and "LISTENING" in l.upper()]
|
||
udp_hit = [l.strip() for l in udp.splitlines() if f":{UDP_PORT}" in l]
|
||
|
||
if tcp_hit:
|
||
for item in tcp_hit:
|
||
line(f" TCP {TCP_PORT} 已被监听： {item}")
|
||
else:
|
||
line(f" TCP {TCP_PORT} 没有监听")
|
||
if udp_hit:
|
||
for item in udp_hit:
|
||
line(f" UDP {UDP_PORT} 已被监听： {item}")
|
||
else:
|
||
line(f" UDP {UDP_PORT} 没有监听")
|
||
|
||
line()
|
||
if agent_running():
|
||
line(" >>> 被控端（ykxmz.exe）正在运行。")
|
||
line(" 它会占着 8765/8766，本自检就绑不上这两个端口。")
|
||
line(" 想用自检抓包：先在它窗口上点「退出」，再重跑本脚本。")
|
||
line(" 其实不用自检也可以 —— 被控端在跑就说明它能收包。")
|
||
else:
|
||
line(" 被控端没有在运行。想连手表的话需要先把它启动。")
|
||
|
||
|
||
def show_firewall() -> None:
|
||
title("3. 防火墙放行状态")
|
||
for name in ("LAN Remote TCP", "LAN Remote UDP", "LAN Remote App TCP"):
|
||
out = run(["netsh", "advfirewall", "firewall", "show", "rule",
|
||
f"name={name}"])
|
||
if not out.strip() or "No rules match" in out or "没有与指定条件" in out:
|
||
line(f" {name}：未放行")
|
||
else:
|
||
enabled = action = ""
|
||
for raw in out.splitlines():
|
||
text = raw.strip()
|
||
if text.startswith("Enabled") or text.startswith("已启用"):
|
||
enabled = text.split(":", 1)[-1].strip()
|
||
elif text.startswith("Action") or text.startswith("操作"):
|
||
action = text.split(":", 1)[-1].strip()
|
||
line(f" {name}：已放行 Enabled={enabled or '?'} Action={action or '?'}")
|
||
|
||
|
||
# ----------------------------------------------------------------------
|
||
# 监听：这段才是关键
|
||
# ----------------------------------------------------------------------
|
||
|
||
received_tcp: list[str] = []
|
||
received_udp: list[str] = []
|
||
bind_errors: list[str] = []
|
||
|
||
|
||
def _bind(sock: socket.socket, port: int) -> bool:
|
||
"""独占绑定。Windows 上必须用 SO_EXCLUSIVEADDRUSE，
|
||
否则第二个进程也能绑上，抓包结果会失真。"""
|
||
if hasattr(socket, "SO_EXCLUSIVEADDRUSE"):
|
||
try:
|
||
sock.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
|
||
except OSError:
|
||
pass
|
||
else:
|
||
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||
try:
|
||
sock.bind(("0.0.0.0", port))
|
||
return True
|
||
except OSError as exc:
|
||
reason = ""
|
||
if exc.errno in _ADDR_IN_USE or (exc.winerror or 0) in _ADDR_IN_USE:
|
||
reason = "端口被占用"
|
||
bind_errors.append(f"{port}: {reason or exc}")
|
||
return False
|
||
|
||
|
||
def tcp_listener(stop: threading.Event) -> None:
|
||
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
if not _bind(server, TCP_PORT):
|
||
line(f" !! 无法监听 TCP {TCP_PORT}")
|
||
if agent_running():
|
||
line(" 原因：被控端正在运行，占着这个端口。")
|
||
line(" -> 先在它窗口上点「退出」，再重跑本自检。")
|
||
return
|
||
server.listen(8)
|
||
server.settimeout(0.5)
|
||
line(f" 正在监听 TCP {TCP_PORT} …")
|
||
while not stop.is_set():
|
||
try:
|
||
conn, addr = server.accept()
|
||
except socket.timeout:
|
||
continue
|
||
except OSError:
|
||
break
|
||
src = f"{addr[0]}:{addr[1]}"
|
||
received_tcp.append(src)
|
||
line(f" [{stamp()}] >>> 收到 TCP 连接：来自 {src} <<<")
|
||
try:
|
||
conn.sendall(b"LANREMOTE-SELFTEST\n")
|
||
except OSError:
|
||
pass
|
||
finally:
|
||
try:
|
||
conn.close()
|
||
except OSError:
|
||
pass
|
||
server.close()
|
||
|
||
|
||
def udp_listener(stop: threading.Event) -> None:
|
||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
if not _bind(sock, UDP_PORT):
|
||
line(f" !! 无法监听 UDP {UDP_PORT}")
|
||
if agent_running():
|
||
line(" 原因：被控端正在运行，占着这个端口。")
|
||
return
|
||
sock.settimeout(0.5)
|
||
line(f" 正在监听 UDP {UDP_PORT} …")
|
||
while not stop.is_set():
|
||
try:
|
||
data, addr = sock.recvfrom(2048)
|
||
except socket.timeout:
|
||
continue
|
||
except OSError:
|
||
break
|
||
src = f"{addr[0]}:{addr[1]}"
|
||
received_udp.append(src)
|
||
preview = data[:80].decode("utf-8", errors="replace")
|
||
line(f" [{stamp()}] >>> 收到 UDP 包：来自 {src} {preview} <<<")
|
||
sock.close()
|
||
|
||
|
||
def run_check(seconds: int = 90) -> int:
|
||
global LISTEN_SECONDS
|
||
LISTEN_SECONDS = int(seconds or 90)
|
||
setup_console()
|
||
|
||
title("局域网遥控 · 网络自检")
|
||
line(" 这个工具监听端口，看手表的请求有没有到达这台电脑。")
|
||
|
||
show_network()
|
||
show_ports()
|
||
show_firewall()
|
||
|
||
title(f"4. 开始监听（{LISTEN_SECONDS} 秒）")
|
||
if agent_running():
|
||
ask_stop_agent()
|
||
line()
|
||
line(" 现在去手表上：打开 App →「设备诊断」→ 底部填上面的 IP → 点「测一下」")
|
||
line()
|
||
|
||
stop = threading.Event()
|
||
for target, args in ((tcp_listener, (stop,)), (udp_listener, (stop,))):
|
||
threading.Thread(target=target, args=args, daemon=True).start()
|
||
time.sleep(0.2)
|
||
|
||
deadline = time.time() + LISTEN_SECONDS
|
||
try:
|
||
while time.time() < deadline:
|
||
time.sleep(0.5)
|
||
left = int(deadline - time.time())
|
||
print(f"\r 剩余 {left:3d} 秒 … 已收到 TCP {len(received_tcp)} 个 / "
|
||
f"UDP {len(received_udp)} 个", end="", flush=True)
|
||
except KeyboardInterrupt:
|
||
pass
|
||
stop.set()
|
||
line()
|
||
line()
|
||
|
||
title("5. 结论")
|
||
if received_tcp or received_udp:
|
||
line(f" TCP 收到 {len(received_tcp)} 个连接，UDP 收到 {len(received_udp)} 个包。")
|
||
line()
|
||
line(" >>> 网络是通的！手表的数据包能到达这台电脑。")
|
||
line(" 既然包能到，连不上就只可能是：")
|
||
line(" * 被控端没在运行")
|
||
line(" * 或端口被别的程序占了")
|
||
elif bind_errors and agent_running():
|
||
line(" 没能抓包 —— 被控端正占着 8765/8766。")
|
||
line()
|
||
line(" >>> 这其实说明被控端是好的。要抓包请先退出它再重跑。")
|
||
else:
|
||
line(" TCP 和 UDP 都一个包都没收到。")
|
||
line()
|
||
line(" >>> 手表的数据包根本没到这台电脑。依次排查：")
|
||
line(" 1) 电脑是不是休眠/关机了")
|
||
line(" 2) 手表连的 WiFi 和电脑是不是同一个")
|
||
line(" 3) 路由器有没有开「AP 隔离 / 客户端隔离」")
|
||
line(" 4) 电脑上是不是有 VPN / 加速器在改路由表")
|
||
line()
|
||
if sys.stdin.isatty():
|
||
line(" 按回车退出…")
|
||
try:
|
||
input()
|
||
except Exception:
|
||
pass
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 直接 `python netcheck.py [秒数]` 也能用。
|
||
# 原来这里写的是 main()，但本文件没有 main() ——
|
||
# 入口是 run_check()，直接跑会 NameError。
|
||
sys.exit(run_check(int(sys.argv[1]) if len(sys.argv) > 1 else 90))
|