mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 14:50:45 +00:00
xmz7891
bf15a2ee61
实测基线：
ykxmz-setup-1.8.1.exe 只下到 237568/15067199 字节就停（17 KB/s）
ykxmz-watch-1.8.1.apk 201523 字节 / 66.8 秒 = 3 KB/s
ykxmz-server-1.8.1.zip 214900 字节 / 3.1 秒 = 67 KB/s
原因：relay 给所有响应都发 Cache-Control: no-store，
Cloudflare 完全不缓存 —— 每次下载/每张图都要穿隧道回 4G 源站。
改动：
/download/* no-store -> public, max-age=604800, immutable
文件名带版本号，同一名字内容永不改变，可以长缓存
/ -> public, max-age=60（挡住刷新重复回源）
/favicon.ico 已是 max-age=86400，保持不变
/shot/* 已是 max-age=86400，保持不变
/v1/* 保持 no-store（接口必须实时）
顺带修两个自己踩的坑：
1. _send() 无条件先发 no-store，再追加 extra 里的 Cache-Control，
结果是**两个 Cache-Control 头**，客户端只认第一个 ——
调用方传的缓存策略被吃掉。改成 extra 里有就不再发默认值。
2. do_HEAD 和 do_GET 的缓存头不一致：HEAD 对非 /shot/ 一律报 no-store。
下载器/代理用 HEAD 探测时会误判文件不可缓存。
现在 _head_info 给每个路由带上和 GET 完全相同的缓存头，
do_HEAD 也不再盲目覆盖。
结果（同一个测试口径）：
setup.exe 17 KB/s（截断） -> 695 KB/s 完整 15 MB 41 倍
watch.apk 3 KB/s -> 101 KB/s 33 倍
server.zip 67 KB/s -> 87 KB/s
三个文件全部 cf-cache-status=HIT（不再穿隧道回源）
排查过程记一笔：我一开始用 curl -I 验证，
看到的全是 no-store —— 因为 **-I 发的是 HEAD**，
而 HEAD 走的是另一个分支。查了半天才发现是验证方法错了。
769 lines
30 KiB
Python
769 lines
30 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""被控端走中转到手表（跨网络）。
|
||
|
||
整体思路
|
||
--------
|
||
现有服务端是"监听 TCP -> accept -> Session(conn) -> 逐条处理命令"。
|
||
跨网络时被控端在 NAT 后面，没法被直连，所以改成**主动连出去**：
|
||
|
||
被控端 --HTTPS 长轮询--> 中转服务器 <--HTTPS 长轮询-- 手表
|
||
|
||
关键技巧：`RelaySocket` 假装成一个 socket。它只实现服务端用到的那几个
|
||
方法（recv / sendall / settimeout / setsockopt / close / shutdown），
|
||
内部把数据压到中转的队列里、从队列里取。
|
||
|
||
于是 `RemoteServer`、`Session`、所有命令处理**一行都不用改** ——
|
||
原来的配对、令牌校验、权限判断全部继续生效。
|
||
中转服务器只是个不认识内容的管道，被攻破也进不了用户的电脑。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
import secrets
|
||
import socket
|
||
import ssl
|
||
import threading
|
||
import time
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
|
||
# 长轮询：中转那边挂 25 秒，客户端超时给宽一点
|
||
# 长轮询单次请求的超时。
|
||
# 必须显著大于服务端的 hold 时间（现在 60 秒），否则会在服务端
|
||
# 正常返回之前自己先超时 —— 那就变成"每次轮询都失败"了。
|
||
# 120 = 60（hold） + 60（网络余量）。
|
||
# 为什么余量给这么大：实测出现过一次 93589ms 的长尾
|
||
# （4G 丢包触发 TCP 重传退避），原来设 90 秒会把它误判成失败。
|
||
POLL_TIMEOUT = 120.0
|
||
# 单次请求体上限（一个 256KB 分块 Base64 后约 350KB，给足余量）
|
||
MAX_BODY = 8 * 1024 * 1024
|
||
# 网络断了之后的退避
|
||
RETRY_MIN = 2.0
|
||
RETRY_MAX = 20.0
|
||
# 多久没成功通信就算掉线（用于界面上显示状态）
|
||
STALE_SECONDS = 90.0
|
||
|
||
|
||
# 请求头。**必须带 User-Agent**：
|
||
# Cloudflare 的 Browser Integrity Check 会把 Python 默认 UA
|
||
# （Python-urllib/3.11）判为可疑客户端，返回 403 + error code 1010。
|
||
# 实测 LanRemote/1.4.0 能过，用自家标识比伪装浏览器合适。
|
||
USER_AGENT = "LanRemote/1.4.0"
|
||
|
||
|
||
# 私有网段 + 回环 + mDNS 名字 —— 这些肯定是局域网，用明文 HTTP
|
||
_PRIVATE_HOST = re.compile(
|
||
r"^(?:"
|
||
r"localhost"
|
||
r"|127\.\d{1,3}\.\d{1,3}\.\d{1,3}"
|
||
r"|10\.\d{1,3}\.\d{1,3}\.\d{1,3}"
|
||
r"|192\.168\.\d{1,3}\.\d{1,3}"
|
||
r"|172\.(?:1[6-9]|2\d|3[01])\.\d{1,3}\.\d{1,3}"
|
||
r"|\d{1,3}(?:\.\d{1,3}){3}" # 任意纯 IP（含公网）
|
||
r"|[^./:]+\.local" # mDNS 名字，如 lanremote.local
|
||
r")"
|
||
r"(?::\d{1,5})?$",
|
||
re.IGNORECASE)
|
||
|
||
|
||
def _guess_scheme(address: str) -> str:
|
||
"""用户没写 http/https 时，猜一个。
|
||
|
||
IP 地址 / localhost / *.local -> http://
|
||
局域网里的中转服务是明文 HTTP。猜成 https 会永远连不上，
|
||
而且报的是 SSL 错误，很容易把人带去查证书（查半天查不出）。
|
||
域名 -> https://
|
||
走公网必须加密（Cloudflare 隧道本来就是 https）。
|
||
"""
|
||
host = address.split("/", 1)[0]
|
||
return "http://" if _PRIVATE_HOST.match(host) else "https://"
|
||
|
||
|
||
|
||
"""用户没写 http/https 时，猜一个。
|
||
|
||
IP 地址 / localhost -> http://
|
||
局域网里的中转服务是明文 HTTP。如果猜成 https，永远连不上，
|
||
而且报的是 SSL 错误，很容易误导人去查证书。
|
||
域名 -> https://
|
||
走公网必须加密（Cloudflare 隧道本来就是 https）。
|
||
"""
|
||
host = address.split("/", 1)[0]
|
||
return "http://" if _PRIVATE_HOST.match(host) else "https://"
|
||
|
||
|
||
def _ssl_context() -> ssl.SSLContext:
|
||
"""给 HTTPS 用的 SSL 上下文。
|
||
|
||
不依赖 certifi：Python 在 Windows 上会从系统证书库加载根证书，
|
||
打包成 EXE 之后也能用。
|
||
"""
|
||
context = ssl.create_default_context()
|
||
try:
|
||
context.load_default_certs()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
return context
|
||
|
||
|
||
class RelayError(Exception):
|
||
pass
|
||
|
||
|
||
class RelaySocket:
|
||
"""假装成一个 socket，实际数据走中转服务器。
|
||
|
||
服务端只用到这几个方法，所以只实现这几个就够了。
|
||
`recv()` 在没有数据时按 `settimeout()` 的设定阻塞等待。
|
||
"""
|
||
|
||
def __init__(self, link: "RelayLink", session_token: str) -> None:
|
||
self._link = link
|
||
self.session_token = session_token
|
||
self._inbox = bytearray()
|
||
self._lock = threading.Lock()
|
||
self._ready = threading.Event()
|
||
self._closed = False
|
||
self._timeout: float | None = None
|
||
self.peer = ("中转服务器", 0)
|
||
|
||
# ---- 服务端会调用的接口 ----
|
||
|
||
def recv(self, size: int) -> bytes:
|
||
deadline = (time.time() + self._timeout) if self._timeout else None
|
||
while True:
|
||
with self._lock:
|
||
if self._inbox:
|
||
data = bytes(self._inbox[:size])
|
||
del self._inbox[:size]
|
||
if not self._inbox:
|
||
self._ready.clear()
|
||
return data
|
||
if self._closed:
|
||
return b""
|
||
if deadline is not None and time.time() >= deadline:
|
||
raise socket.timeout("等待中转数据超时")
|
||
self._ready.wait(0.5)
|
||
if self._closed:
|
||
return b""
|
||
|
||
def sendall(self, data: bytes) -> None:
|
||
if self._closed:
|
||
raise OSError("中转连接已关闭")
|
||
self._link.send_to_client(data)
|
||
|
||
def settimeout(self, value) -> None:
|
||
self._timeout = value
|
||
|
||
def setsockopt(self, *_args, **_kwargs) -> None:
|
||
"""TCP_NODELAY 之类，这里没有意义，忽略掉就行。"""
|
||
|
||
def getpeername(self):
|
||
return self.peer
|
||
|
||
def shutdown(self, _how=0) -> None:
|
||
self.close()
|
||
|
||
def close(self) -> None:
|
||
self._closed = True
|
||
self._ready.set()
|
||
|
||
def fileno(self) -> int:
|
||
# 有些代码会拿它做 select；这里不支持，返回 -1 明确表示
|
||
return -1
|
||
|
||
# ---- 中转那边把数据喂进来 ----
|
||
|
||
def feed(self, data: bytes) -> None:
|
||
if self._closed or not data:
|
||
return
|
||
with self._lock:
|
||
self._inbox.extend(data)
|
||
self._ready.set()
|
||
|
||
@property
|
||
def is_closed(self) -> bool:
|
||
return self._closed
|
||
|
||
|
||
class RelayLink:
|
||
"""被控端到中转服务器的长连接。
|
||
|
||
一条链路，一个"当前手表会话"。中转那边一台设备同时只允许一个
|
||
手表会话，所以这里也只需要维护一个 RelaySocket。
|
||
"""
|
||
|
||
def __init__(self, store, log=None, on_session=None, on_alert=None) -> None:
|
||
self.store = store
|
||
self._log = log or (lambda _m: None)
|
||
self._on_session = on_session
|
||
# 出需要用户处理的事（比如配对码被占）时调这个，用来弹通知
|
||
self._on_alert = on_alert
|
||
|
||
self.url = ""
|
||
self.candidates: list = []
|
||
self._fail_count = 0
|
||
self.device_id = ""
|
||
self.device_name = str(store.get("server_name") or "")
|
||
self.token = ""
|
||
self.code = ""
|
||
self.online = False
|
||
self.last_ok = 0.0
|
||
self.last_error = ""
|
||
self.sessions_served = 0
|
||
# 配对码冲突的说明。非空表示现在卡在这，等用户换码。
|
||
self.code_conflict = ""
|
||
self._alerted_code = ""
|
||
|
||
self._thread: threading.Thread | None = None
|
||
self._running = False
|
||
self._socket: RelaySocket | None = None
|
||
self._socket_lock = threading.Lock()
|
||
self._retry = RETRY_MIN
|
||
# 复用的 HTTP 连接。
|
||
# 为什么必须复用：这条链路丢包约 40%，每次新建 TLS 连接有约
|
||
# 30% 概率握手失败；连接建好后后续请求就 100% 成功。
|
||
# 实测：每次新建 70% 可用，复用 100%。
|
||
self._http = None
|
||
self._http_lock = threading.RLock()
|
||
|
||
# ------------------------------------------------------------------
|
||
# 对外
|
||
# ------------------------------------------------------------------
|
||
|
||
def start(self, url: str) -> bool:
|
||
"""开始连中转。
|
||
|
||
url 支持用**分号**分隔多个地址，按顺序尝试：
|
||
http://192.168.2.156:8080;https://relay.xxmxxs.top
|
||
|
||
为什么需要多个：
|
||
* 电脑和随身 WiFi 常在同一局域网 —— 直连局域网中转更快更稳，
|
||
而且地址永远不变（不用管隧道地址变来变去）
|
||
* 电脑在外面时才走隧道
|
||
"""
|
||
candidates = []
|
||
for piece in (url or "").replace(",", ";").split(";"):
|
||
piece = piece.strip().rstrip("/")
|
||
if not piece:
|
||
continue
|
||
if not piece.startswith(("http://", "https://")):
|
||
piece = _guess_scheme(piece) + piece
|
||
if piece not in candidates:
|
||
candidates.append(piece)
|
||
|
||
if not candidates:
|
||
self._log("没配置中转地址，跳过")
|
||
return False
|
||
|
||
# 上次连通的地址排到最前面。
|
||
# 设备 IP 会变（DHCP），但大概率还是上次那个；
|
||
# 排前面能省掉"挨个探到超时"的那几秒（实测 2 个候选要 10 秒，
|
||
# 而这段时间局域网发现都还没起来，手表搜不到电脑）。
|
||
try:
|
||
last_good = str(self.store.get("relay_last_good") or "").strip()
|
||
except Exception: # noqa: BLE001
|
||
last_good = ""
|
||
if last_good:
|
||
normalized = last_good.rstrip("/")
|
||
if normalized in candidates:
|
||
candidates.remove(normalized)
|
||
candidates.insert(0, normalized)
|
||
|
||
self.candidates = candidates
|
||
# 先探一遍，挑第一个通的
|
||
chosen = self._pick_candidate(candidates)
|
||
self.url = chosen
|
||
if len(candidates) > 1:
|
||
self._log(f"中转地址候选 {len(candidates)} 个，选用 {chosen}")
|
||
|
||
# 记下来，下次优先试它
|
||
if chosen:
|
||
try:
|
||
if str(self.store.get("relay_last_good") or "") != chosen:
|
||
self.store.set("relay_last_good", chosen)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
self.device_id = self._ensure_device_id()
|
||
self.code = str(self.store.pairing_code() or "")
|
||
|
||
self._running = True
|
||
self._thread = threading.Thread(target=self._run, daemon=True,
|
||
name="relay-link")
|
||
self._thread.start()
|
||
self._log(f"中转已启用：{self.url}")
|
||
return True
|
||
|
||
def stop(self) -> None:
|
||
self._running = False
|
||
with self._socket_lock:
|
||
if self._socket is not None:
|
||
self._socket.close()
|
||
self._socket = None
|
||
if self.token:
|
||
try:
|
||
self._post("/v1/agent/bye", b"", {"X-Token": self.token},
|
||
timeout=5)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
def status(self) -> dict:
|
||
return {
|
||
"enabled": bool(self.url),
|
||
"url": self.url,
|
||
"candidates": list(self.candidates),
|
||
"codeConflict": self.code_conflict,
|
||
"deviceId": self.device_id,
|
||
"code": self.code,
|
||
"online": self.online and (time.time() - self.last_ok < STALE_SECONDS),
|
||
"idleSeconds": int(time.time() - self.last_ok) if self.last_ok else -1,
|
||
"sessions": self.sessions_served,
|
||
"error": self.last_error,
|
||
}
|
||
|
||
def send_to_client(self, data: bytes) -> None:
|
||
"""把数据发给当前手表。"""
|
||
if not self.token:
|
||
raise OSError("还没连上中转")
|
||
if not data:
|
||
return
|
||
for start in range(0, len(data), 256 * 1024):
|
||
piece = data[start:start + 256 * 1024]
|
||
self._post("/v1/agent/send", piece, {"X-Token": self.token},
|
||
timeout=POLL_TIMEOUT)
|
||
|
||
# ------------------------------------------------------------------
|
||
# 内部
|
||
# ------------------------------------------------------------------
|
||
|
||
def _probe(self, base: str, timeout: float = 8.0) -> bool:
|
||
"""这个地址通不通（打一下 /v1/health）。"""
|
||
try:
|
||
request = urllib.request.Request(
|
||
base + "/v1/health", headers={"User-Agent": USER_AGENT})
|
||
context = _ssl_context() if base.startswith("https://") else None
|
||
with urllib.request.urlopen(request, timeout=timeout,
|
||
context=context) as response:
|
||
if response.status != 200:
|
||
return False
|
||
body = response.read(4096).decode("utf-8", "replace")
|
||
return '"ok"' in body
|
||
except Exception: # noqa: BLE001
|
||
return False
|
||
|
||
def _pick_candidate(self, candidates: list) -> str:
|
||
for base in candidates:
|
||
if self._probe(base):
|
||
return base
|
||
self._log(f" {base} 不通，试下一个")
|
||
# 都不通也先返回第一个，让主循环去重试
|
||
return candidates[0]
|
||
|
||
def _prefer_lan(self) -> None:
|
||
"""定期回头看看：局域网那条路是不是又通了？
|
||
|
||
为什么需要：
|
||
实测局域网中转 3.5ms、公网隧道 1.5 秒（还只有 75% 可用）。
|
||
原来的逻辑只在**报错时**才切换地址，
|
||
所以一旦切到隧道，就算局域网恢复了也不会切回来 ——
|
||
白背着 400 倍的延迟和 25% 的丢包。
|
||
|
||
策略：每 60 秒挑一个"比当前更靠前"的候选试一下
|
||
（候选表是按优先级排的：局域网在前，隧道在后），
|
||
通了就切回去。只往前切、不往后切，避免来回抖。
|
||
"""
|
||
if not self.candidates or not self.url:
|
||
return
|
||
try:
|
||
current = self.candidates.index(self.url)
|
||
except ValueError:
|
||
current = len(self.candidates)
|
||
# 只试排在当前前面的那些（优先级更高）
|
||
for base in self.candidates[:current]:
|
||
if self._probe(base, timeout=3):
|
||
self._log(f"发现更好的中转地址，切回 {base}")
|
||
self.url = base
|
||
self._drop_http() # 换地址了，旧连接作废
|
||
return
|
||
|
||
def _refresh_url(self) -> None:
|
||
"""主地址连不上时，去候选里找找有没有能用的。
|
||
|
||
如果某个候选本身就是局域网中转，还能顺便问它
|
||
「现在的隧道地址是什么」（/url 接口），自动跟随地址变化。
|
||
"""
|
||
for base in list(self.candidates):
|
||
if base == self.url:
|
||
continue
|
||
if not self._probe(base, timeout=5):
|
||
continue
|
||
self._log(f"切换到备用中转地址：{base}")
|
||
self.url = base
|
||
return
|
||
|
||
def _lan_ips(self) -> list:
|
||
"""本机在局域网里的 IP（可能有多张网卡）。
|
||
|
||
手表靠这个判断"电脑是不是和我同一个网络"。
|
||
拿不到就返回空列表 —— 中转那边会当成"不知道"，
|
||
手表就继续走中转，不会出错。
|
||
"""
|
||
found = []
|
||
try:
|
||
probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
try:
|
||
# 不会真的发包，只是让系统选出默认出口网卡
|
||
probe.connect(("8.8.8.8", 80))
|
||
addr = probe.getsockname()[0]
|
||
if addr and not addr.startswith("127."):
|
||
found.append(addr)
|
||
finally:
|
||
probe.close()
|
||
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 not addr.startswith("169.254.")
|
||
and addr not in found):
|
||
found.append(addr)
|
||
except OSError:
|
||
pass
|
||
return found
|
||
|
||
def _check_code(self, code: str) -> dict:
|
||
"""问中转：这个码被占了吗？（只读，不会占用它）
|
||
|
||
返回 {"ok": True, "taken": bool, "deviceName": str}
|
||
连不上中转就返回 {"ok": False}，调用方自己决定怎么办。
|
||
"""
|
||
bases = list(self.candidates) or ([self.url] if self.url else [])
|
||
if not bases:
|
||
return {"ok": False, "error": "还没配置中转地址"}
|
||
quoted = urllib.parse.quote(code)
|
||
for base in bases:
|
||
try:
|
||
request = urllib.request.Request(
|
||
f"{base}/v1/code/check?code={quoted}",
|
||
headers={"User-Agent": USER_AGENT})
|
||
context = (_ssl_context() if base.startswith("https://")
|
||
else None)
|
||
with urllib.request.urlopen(request, timeout=10,
|
||
context=context) as response:
|
||
return json.loads(response.read(4096).decode(
|
||
"utf-8", "replace"))
|
||
except Exception as exc: # noqa: BLE001
|
||
last = f"{type(exc).__name__}: {exc}"
|
||
continue
|
||
return {"ok": False, "error": last if "last" in dir() else "连不上中转"}
|
||
|
||
def _ensure_device_id(self) -> str:
|
||
"""设备唯一标识。存下来，重装/重启都不变 —— 单设备单配对码要用它。"""
|
||
existing = str(self.store.get("relay_device_id") or "").strip()
|
||
if existing:
|
||
return existing
|
||
# 带上机器名，便于在服务器管理页上认出来
|
||
name = str(self.store.get("server_name") or socket.gethostname())
|
||
safe = "".join(ch for ch in name if ch.isalnum() or ch in "-_")[:24]
|
||
device_id = f"{safe or 'pc'}-{secrets.token_hex(6)}"
|
||
self.store.set("relay_device_id", device_id)
|
||
return device_id
|
||
|
||
def _run(self) -> None:
|
||
backoff = RETRY_MIN
|
||
while self._running:
|
||
try:
|
||
if not self.token:
|
||
self._register()
|
||
self._poll_once()
|
||
backoff = RETRY_MIN
|
||
# 每分钟回头看看有没有更快的中转（局域网优先）
|
||
now = time.time()
|
||
if now - getattr(self, "_last_prefer", 0.0) >= 60.0:
|
||
self._last_prefer = now
|
||
self._prefer_lan()
|
||
except _AuthLost:
|
||
self._log("中转说令牌失效，重新登记")
|
||
self.token = ""
|
||
time.sleep(1)
|
||
except RelayCodeTaken as exc:
|
||
# 配对码被别的设备占了。这不是网络问题，重试没用。
|
||
# 弹一次通知，然后**停下来等用户换码**。
|
||
message = str(exc) or "该配对码已被使用，请换一个配对码"
|
||
self.online = False
|
||
self.token = ""
|
||
self.last_error = message
|
||
self.code_conflict = message
|
||
current = self.store.pairing_code()
|
||
if current != self._alerted_code:
|
||
# 只在换了新码之后再弹，避免反复弹同一个码的通知
|
||
self._alerted_code = current
|
||
self._log(f"配对码冲突：{message}")
|
||
if self._on_alert:
|
||
try:
|
||
self._on_alert("配对码被占用", message)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
# 等用户换码。每 5 秒看一次配对码变没变。
|
||
for _ in range(12):
|
||
if not self._running:
|
||
return
|
||
time.sleep(0.5)
|
||
if self.store.pairing_code() != current:
|
||
self._log("配对码变了，重新登记")
|
||
self._alerted_code = ""
|
||
break
|
||
continue
|
||
except Exception as exc: # noqa: BLE001
|
||
self.online = False
|
||
self.last_error = f"{type(exc).__name__}: {exc}"
|
||
self._log(f"中转连接出问题：{self.last_error}")
|
||
# 网络断了：把当前会话关掉，让上层重开
|
||
self._drop_socket()
|
||
self.token = ""
|
||
# 连续失败几次就换一个候选地址试试
|
||
fail_count = getattr(self, "_fail_count", 0) + 1
|
||
self._fail_count = fail_count
|
||
if fail_count >= 3 and len(getattr(self, "candidates", [])) > 1:
|
||
self._fail_count = 0
|
||
self._refresh_url()
|
||
time.sleep(backoff)
|
||
backoff = min(RETRY_MAX, backoff * 1.6)
|
||
|
||
def _register(self) -> None:
|
||
# 每次登记都从设置里重新读配对码。
|
||
# 用户可能刚改过（比如上一个码被别的设备占了），
|
||
# 不重读就会一直拿旧码去报，永远卡在冲突里出不来。
|
||
latest = str(self.store.pairing_code() or "").strip()
|
||
if latest and latest != self.code:
|
||
self._log(f"配对码已改为 {latest}，用新的重新登记")
|
||
self.code = latest
|
||
|
||
try:
|
||
listen_port = int(self.store.get("port") or 0)
|
||
except (TypeError, ValueError):
|
||
listen_port = 0
|
||
payload = json.dumps({
|
||
"deviceId": self.device_id,
|
||
"code": self.code,
|
||
"name": self.device_name or self.device_id,
|
||
"version": _version(),
|
||
# 手表用这两个判断要不要切成局域网直连
|
||
"lanIps": self._lan_ips(),
|
||
"port": listen_port,
|
||
}, ensure_ascii=False).encode("utf-8")
|
||
try:
|
||
status, body = self._post("/v1/agent/register", payload, {},
|
||
timeout=30)
|
||
except urllib.error.HTTPError as exc:
|
||
# 中转用 400 表示"登记被拒"，错误详情在响应体里。
|
||
# urlopen 抛 HTTPError 时不会把 body 交出来，得自己读。
|
||
# 不读的话配对码冲突就被当成普通网络错误，
|
||
# 用户永远看不到"该配对码已被使用"的通知。
|
||
body = exc.read(MAX_BODY)
|
||
try:
|
||
reply = json.loads(body.decode("utf-8", "replace"))
|
||
except Exception: # noqa: BLE001
|
||
raise RelayError(f"登记响应看不懂：{body[:120]!r}") from None
|
||
if not reply.get("ok"):
|
||
message = str(reply.get("error") or "登记被拒")
|
||
# 配对码被占：精确判断（服务端会带 codeTaken），
|
||
# 兼容老服务端就匹配关键字。
|
||
if reply.get("codeTaken") or "已被" in message and "配对码" in message:
|
||
raise RelayCodeTaken(message)
|
||
raise RelayError(message)
|
||
# 登记成功 -> 之前的冲突解除了
|
||
self.code_conflict = ""
|
||
|
||
self.token = str(reply.get("token") or "")
|
||
# 中转可能按「单设备单配对码」把码改回旧的，这里跟着改
|
||
server_code = str(reply.get("code") or "")
|
||
if server_code and server_code != self.code:
|
||
self.code = server_code
|
||
try:
|
||
self.store.set_pairing_code(server_code)
|
||
self._log(f"中转按单设备单配对码把配对码改回 {server_code}")
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
note = str(reply.get("note") or "")
|
||
if note:
|
||
self._log(note)
|
||
self.online = True
|
||
self.last_ok = time.time()
|
||
self.last_error = ""
|
||
self._log(f"已登记到中转：{self.device_name} 配对码 {self.code}")
|
||
|
||
def _poll_once(self) -> None:
|
||
headers = {"X-Token": self.token}
|
||
try:
|
||
status, body, extra = self._post(
|
||
"/v1/agent/poll", b"", headers, timeout=POLL_TIMEOUT,
|
||
want_headers=True)
|
||
except urllib.error.HTTPError as exc:
|
||
if exc.code == 401:
|
||
raise _AuthLost() from None
|
||
raise
|
||
|
||
self.online = True
|
||
self.last_ok = time.time()
|
||
self.last_error = ""
|
||
|
||
# 会话变了 -> 手表换了一台/重连了 -> 重开一条会话
|
||
session_token = (extra.get("X-Relay-Session") or "-").strip()
|
||
self._sync_session(session_token)
|
||
|
||
if body and session_token not in ("", "-"):
|
||
with self._socket_lock:
|
||
if self._socket is not None:
|
||
self._socket.feed(body)
|
||
|
||
def _sync_session(self, session_token: str) -> None:
|
||
"""中转报的会话令牌和本地不一致时，重建 RelaySocket。"""
|
||
with self._socket_lock:
|
||
current = self._socket
|
||
if current is not None and current.session_token == session_token:
|
||
return
|
||
if session_token in ("", "-"):
|
||
# 手表都不在了，把当前会话关掉
|
||
if current is not None:
|
||
current.close()
|
||
self._socket = None
|
||
return
|
||
if current is not None:
|
||
current.close()
|
||
new_socket = RelaySocket(self, session_token)
|
||
self._socket = new_socket
|
||
self.sessions_served += 1
|
||
|
||
self._log(f"手表接入中转（会话 {session_token[:8]}…）")
|
||
if self._on_session is not None:
|
||
try:
|
||
self._on_session(new_socket)
|
||
except Exception as exc: # noqa: BLE001
|
||
self._log(f"建立会话失败：{exc}")
|
||
|
||
def _drop_socket(self) -> None:
|
||
with self._socket_lock:
|
||
if self._socket is not None:
|
||
self._socket.close()
|
||
self._socket = None
|
||
|
||
# ---- HTTP ----
|
||
|
||
def _open_http(self):
|
||
"""按 self.url 建一条 HTTP(S) 连接。"""
|
||
import http.client
|
||
parts = urllib.parse.urlsplit(self.url)
|
||
host = parts.hostname or ""
|
||
if parts.scheme == "https":
|
||
return http.client.HTTPSConnection(
|
||
host, parts.port or 443, timeout=POLL_TIMEOUT,
|
||
context=_ssl_context())
|
||
return http.client.HTTPConnection(host, parts.port or 80,
|
||
timeout=POLL_TIMEOUT)
|
||
|
||
def _drop_http(self) -> None:
|
||
"""把缓存的连接关掉（出错后调用，下次会重开）。"""
|
||
with self._http_lock:
|
||
connection, self._http = self._http, None
|
||
if connection is not None:
|
||
try:
|
||
connection.close()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
def _post(self, path: str, body: bytes, headers: dict,
|
||
timeout: float = POLL_TIMEOUT, want_headers: bool = False):
|
||
"""POST 一个请求，**复用同一条连接**。
|
||
|
||
为什么要复用（真机实测）：
|
||
这条链路丢包约 40%，每次新建 TLS 连接约 30% 概率握手失败。
|
||
但连接建好之后，后续请求 100% 成功（TCP 会自己重传）。
|
||
实测：每次新建 -> 70% 可用；复用 -> 100%。
|
||
|
||
所以这里：
|
||
* 优先用缓存的那条连接
|
||
* 断了就重开一次再试（短抖动自己扛过去）
|
||
* 两次都失败才算真失败，往上抛
|
||
"""
|
||
send = dict(headers or {})
|
||
send["Content-Type"] = "application/octet-stream"
|
||
send["User-Agent"] = "LanRemote-Agent/" + _version()
|
||
send["Connection"] = "keep-alive"
|
||
|
||
last_error = None
|
||
for attempt in (1, 2):
|
||
with self._http_lock:
|
||
connection = self._http
|
||
if connection is None:
|
||
connection = self._open_http()
|
||
self._http = connection
|
||
try:
|
||
connection.timeout = timeout
|
||
connection.request("POST", path, body=body, headers=send)
|
||
response = connection.getresponse()
|
||
payload = response.read(MAX_BODY)
|
||
status = response.status
|
||
# 4xx/5xx：抛 HTTPError，但**必须把响应体带上**。
|
||
# 中转用 400 + JSON body 表示"登记被拒"（配对码被占），
|
||
# 上层要读这个 body 才能区分"码被占"和"普通网络错误"。
|
||
# 早先这里传 fp=None，导致 exc.read() 永远是 b''，
|
||
# 配对码冲突就识别不出来了。
|
||
if status >= 400:
|
||
import io as _io
|
||
self._drop_http()
|
||
reason = "unauthorized" if status == 401 else "http error"
|
||
raise urllib.error.HTTPError(
|
||
self.url + path, status, reason,
|
||
response.headers, _io.BytesIO(payload))
|
||
if want_headers:
|
||
extra = {name: value
|
||
for name, value in response.headers.items()}
|
||
return status, payload, extra
|
||
return status, payload
|
||
except urllib.error.HTTPError:
|
||
raise # HTTP 状态码是业务层的，不算连接问题
|
||
except Exception as exc: # noqa: BLE001
|
||
# 连接层面的问题：关掉，下一轮重开
|
||
last_error = exc
|
||
self._drop_http()
|
||
if attempt == 2:
|
||
break
|
||
self._log_quiet(f"连接断了，重开一次再试（{exc}）")
|
||
|
||
raise RelayError(f"请求失败: {last_error}")
|
||
|
||
def _log_quiet(self, message: str) -> None:
|
||
"""记一条不打扰用户的日志（重连这种事不用弹通知）。"""
|
||
try:
|
||
self._log(message)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
|
||
class RelayCodeTaken(RelayError):
|
||
"""配对码被别的设备占了。
|
||
|
||
这**不是网络问题**，重试没用 —— 要等用户换一个配对码。
|
||
所以主循环遇到它不能当成普通错误一直重试（会刷屏），
|
||
而是弹一次通知然后停下等。
|
||
"""
|
||
|
||
|
||
class _AuthLost(Exception):
|
||
"""中转不认识我们的令牌了（重启过 / 被清理），要重新登记。"""
|
||
|
||
|
||
def _version() -> str:
|
||
try:
|
||
from . import __version__
|
||
return __version__
|
||
except Exception: # noqa: BLE001
|
||
return "1.4.0"
|