mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 14:50:45 +00:00
xmz7891
566d9e384f
阶段1 网络加速 - 发现隧道在跑 QUIC（ExecStart 丢了 --protocol http2）， 实测 QUIC 可用率 48% vs HTTP/2 80%+，是重大性能回退。 恢复后公网延迟从 4141ms 降到 690ms（快 6 倍），可用率 92%。 - 优选 IP 结论：无效。hosts 覆盖确实生效了（日志里连的就是指定 IP）， 但落点仍是 lax —— 运营商把所有 Cloudflare anycast 都路由到美西。 已清掉这段无效配置。 - BBR 不可用：设备是自定义内核 6.4.0-rc4-jsbsbxjxh66， 没编 tcp_bbr 模块也没配置项。改用 cubic + 调优： TCP 缓冲 4MB、tcp_fastopen=3、tcp_slow_start_after_idle=0 （长轮询场景关键）、tcp_tw_reuse=1。 阶段2 官网重做 - 拆出独立模块 relay/landing.py（relay.py 已两千多行， 再塞 HTML 没法看） - 信息结构参考成熟产品站：吸顶导航 / Hero / 九宫格功能 / 三步上手 / 下载 / FAQ / 页脚 - 保持白色简洁：底白、浅灰面板、细边框、大量留白 - logo 用内联 SVG（ICO 放 img 里实测是破图） - 管理入口从官网移除（管理不对外开放） 过程中被自己的测试抓到两个 bug： - 模块叫 site.py 会和 Python 标准库重名，import site 拿到的是标准库 - _logo() 里把 % 格式化和字符串拼接混用，直接抛 TypeError
246 lines
8.7 KiB
Python
246 lines
8.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""验证静默模式。
|
|
|
|
三件事必须成立：
|
|
1. 启动后**一个窗口都没有** —— 用 EnumWindows 枚举该进程的所有顶层窗口来证明，
|
|
不是"看起来没有"
|
|
2. 真的在监听，手表能连上
|
|
3. **配对认证仍然生效** —— 静默模式默认不接受新设备，
|
|
否则局域网里任何人都能接进来，那就成了后门
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import ctypes
|
|
import json
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
AGENT = ROOT / "windows" / "run_agent.py"
|
|
DATA = ROOT / "windows" / "data"
|
|
PORT = 8877
|
|
|
|
PASSED = 0
|
|
FAILED = 0
|
|
|
|
|
|
def check(name, ok, detail=""):
|
|
global PASSED, FAILED
|
|
if ok:
|
|
PASSED += 1
|
|
print(f" [PASS] {name}")
|
|
else:
|
|
FAILED += 1
|
|
print(f" [FAIL] {name} {detail}")
|
|
|
|
|
|
def windows_of_process(pid: int) -> list:
|
|
"""枚举属于该进程的所有可见/不可见顶层窗口。"""
|
|
user32 = ctypes.windll.user32
|
|
found: list = []
|
|
EnumProc = ctypes.WINFUNCTYPE(ctypes.c_bool, ctypes.c_void_p,
|
|
ctypes.c_void_p)
|
|
|
|
def callback(hwnd, _lparam):
|
|
owner = ctypes.c_ulong()
|
|
user32.GetWindowThreadProcessId(hwnd, ctypes.byref(owner))
|
|
if owner.value == pid:
|
|
length = user32.GetWindowTextLengthW(hwnd)
|
|
buffer = ctypes.create_unicode_buffer(max(2, length + 1))
|
|
user32.GetWindowTextW(hwnd, buffer, length + 1)
|
|
cls = ctypes.create_unicode_buffer(256)
|
|
user32.GetClassNameW(hwnd, cls, 256)
|
|
found.append({
|
|
"title": buffer.value,
|
|
"class": cls.value,
|
|
"visible": bool(user32.IsWindowVisible(hwnd)),
|
|
})
|
|
return True
|
|
|
|
user32.EnumWindows(EnumProc(callback), 0)
|
|
return found
|
|
|
|
|
|
def port_open(port: int) -> bool:
|
|
try:
|
|
sock = socket.create_connection(("127.0.0.1", port), timeout=1)
|
|
sock.close()
|
|
return True
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
|
|
def seed_paired_device(data_dir):
|
|
"""预置一台已配设备。
|
|
|
|
现在的规则是：**一台设备都没配对过时，配对是开着的**
|
|
（这是修用户报的"设完码却连不上"）。所以想测
|
|
"静默模式拒绝新配对"，得先让电脑认为已经有设备配过了。
|
|
"""
|
|
import json as _json
|
|
data_dir = Path(data_dir)
|
|
data_dir.mkdir(parents=True, exist_ok=True)
|
|
(data_dir / "devices.json").write_text(
|
|
_json.dumps({"already-paired": {
|
|
"id": "already-paired", "name": "已配的手表",
|
|
"token": "seed-token", "last": 0}}, ensure_ascii=False),
|
|
encoding="utf-8")
|
|
|
|
|
|
def seed_setup_done(data_dir):
|
|
"""预先置 setup_done：这些测试测的是别的功能，不该弹出首次设置窗口。"""
|
|
import json as _json
|
|
data_dir.mkdir(parents=True, exist_ok=True)
|
|
settings = data_dir / "settings.json"
|
|
payload = {}
|
|
if settings.is_file():
|
|
try:
|
|
payload = _json.loads(settings.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
payload = {}
|
|
payload["setup_done"] = True
|
|
settings.write_text(_json.dumps(payload, ensure_ascii=False, indent=2),
|
|
encoding="utf-8")
|
|
|
|
def main() -> int:
|
|
backup = None
|
|
if DATA.exists():
|
|
backup = DATA.parent / f"data_backup_silent_{int(time.time())}"
|
|
shutil.move(str(DATA), str(backup))
|
|
print(f" 已备份真实 data")
|
|
|
|
seed_setup_done(DATA)
|
|
seed_paired_device(DATA)
|
|
proc = None
|
|
try:
|
|
proc = subprocess.Popen(
|
|
[sys.executable, str(AGENT), "--port", str(PORT), "--silent"],
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
print(f" 已静默启动（PID {proc.pid}），等待监听…")
|
|
|
|
listening = False
|
|
for _ in range(30):
|
|
time.sleep(0.5)
|
|
if port_open(PORT):
|
|
listening = True
|
|
break
|
|
|
|
check("静默模式起来了并在监听", listening)
|
|
|
|
print()
|
|
print("一、窗口检查（EnumWindows 枚举该进程的顶层窗口）")
|
|
time.sleep(1.5)
|
|
windows = windows_of_process(proc.pid)
|
|
visible = [w for w in windows if w["visible"]]
|
|
# Tk 会创建一个隐藏的辅助窗口，那不算"界面窗口"
|
|
ui_windows = [w for w in visible if w["title"]]
|
|
print(f" 该进程顶层窗口 {len(windows)} 个，"
|
|
f"可见 {len(visible)} 个，可见且有标题 {len(ui_windows)} 个")
|
|
for w in windows[:6]:
|
|
print(f" class={w['class']!r} title={w['title']!r} "
|
|
f"visible={w['visible']}")
|
|
check("没有任何可见窗口", not visible,
|
|
f"竟然有 {len(visible)} 个: {visible[:3]}")
|
|
check("没有带标题的界面窗口", not ui_windows, str(ui_windows[:3]))
|
|
|
|
print()
|
|
print("二、终端检查")
|
|
check("启动时没有附带控制台窗口",
|
|
not any(w["class"] == "ConsoleWindowClass" for w in windows))
|
|
|
|
print()
|
|
print("三、★ 配对认证必须仍然生效")
|
|
settings = DATA / "settings.json"
|
|
code = ""
|
|
if settings.is_file():
|
|
try:
|
|
code = json.loads(settings.read_text(encoding="utf-8")).get(
|
|
"pairing_code", "")
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
print(f" 配对码 {code or '(无)'}，静默模式默认不接受新设备")
|
|
|
|
# 故意用一个**错误的码** —— 安全边界是"码错必须被拒"。
|
|
# （以前这里拿正确的码试也期待被拒，但后来改成
|
|
# "配对码本身就是密钥，码对就放行"，所以那样测就不对了。）
|
|
wrong = "000000" if code != "000000" else "111111"
|
|
|
|
sock = socket.create_connection(("127.0.0.1", PORT), timeout=5)
|
|
sock.settimeout(8)
|
|
request = {"cmd": "pair", "deviceId": "silent-test",
|
|
"deviceName": "测试", "code": wrong}
|
|
sock.sendall((json.dumps(request, ensure_ascii=False) + "\n").encode("utf-8"))
|
|
raw = b""
|
|
while b"\n" not in raw:
|
|
chunk = sock.recv(4096)
|
|
if not chunk:
|
|
break
|
|
raw += chunk
|
|
reply = json.loads(raw.split(b"\n")[0].decode("utf-8")) if raw else {}
|
|
print(f" 配对回复: {reply}")
|
|
# 行为变了：配对窗口关着时，**码错才拒**，码对放行。
|
|
# 因为配对码本身就是密钥，老逻辑"首台配对后自动关窗"
|
|
# 会让重装手表的人永久连不上（实测踩过）。
|
|
check("静默模式下拿着错误的码配不上",
|
|
reply.get("ok") is False,
|
|
f"竟然配上了: {reply}")
|
|
check("给出的理由是明确的",
|
|
"配对" in str(reply.get("error", "")) or
|
|
"关闭" in str(reply.get("error", "")),
|
|
str(reply.get("error")))
|
|
|
|
print()
|
|
print("四、未配对的设备不能执行任何命令")
|
|
sock.sendall((json.dumps({"cmd": "feature_list"},
|
|
ensure_ascii=False) + "\n").encode("utf-8"))
|
|
raw = b""
|
|
while b"\n" not in raw:
|
|
chunk = sock.recv(4096)
|
|
if not chunk:
|
|
break
|
|
raw += chunk
|
|
reply = json.loads(raw.split(b"\n")[0].decode("utf-8")) if raw else {}
|
|
check("未配对时功能列表被拒绝", reply.get("ok") is False,
|
|
f"竟然返回了: {str(reply)[:80]}")
|
|
sock.close()
|
|
|
|
print()
|
|
print("五、日志文件")
|
|
log = DATA / "silent.log"
|
|
check("静默模式写了日志", log.is_file(), str(log))
|
|
|
|
finally:
|
|
if proc is not None:
|
|
try:
|
|
proc.terminate()
|
|
proc.wait(timeout=8)
|
|
except Exception: # noqa: BLE001
|
|
try:
|
|
proc.kill()
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
time.sleep(1.0)
|
|
try:
|
|
if DATA.exists():
|
|
shutil.rmtree(DATA, ignore_errors=True)
|
|
if backup is not None and backup.exists():
|
|
shutil.move(str(backup), str(DATA))
|
|
print(" 已还原真实 data")
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f" !! 还原失败: {exc}")
|
|
|
|
print()
|
|
print("=" * 62)
|
|
print(f"通过 {PASSED} 项，失败 {FAILED} 项")
|
|
return 1 if FAILED else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|