ykxmz/windows/lanremote/relay.py

# -*- 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 秒），否则会在服务端
#   正常返回之前自己先超时 —— 那就变成"每次轮询都失败"了。
#   180 = 60（hold） + 120（网络余量）。
#   余量一直在加大，因为实测的长尾一直在刷新：
#     90 秒 -> 实测 93589ms  不够
#    120 秒 -> 实测 126604ms 不够（30 次里出现 1 次，其余都是 61~62 秒）
#   4G 丢包触发的重传退避能拖出一分钟以上的停顿，这是链路特性。
#   放宽超时只是"别把慢但成功的请求冤枉掉"，不增加请求数。
#   而且客户端有"连续失败 3 次才判离线"的保护，
#   偶尔一次长等待不会让界面闪"已断开"。
POLL_TIMEOUT = 180.0
# 单次请求体上限（一个 256KB 分块 Base64 后约 350KB，给足余量）
MAX_BODY = 8 * 1024 * 1024
# 网络断了之后的退避
RETRY_MIN = 2.0
RETRY_MAX = 20.0
# 连续失败几次才判定"离线"。
#   单次失败就报离线太敏感：4G 链路上有 2~4% 的偶发失败，
#   而链路通常一两秒就自己恢复，界面却已经闪了一下"已断开"。
#   连续 3 次（至少 6 秒都不通）才判离线，真的断了仍然如实显示。
OFFLINE_AFTER = 3
# 多久没成功通信就算掉线（用于界面上显示状态）
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()
        # 缓存的那条连接现在有没有在途请求。
        #   /v1/agent/poll 是长轮询，一条请求挂 180 秒 ——
        #   那段时间里这条连接**不能**被别的线程再拿去发请求，
        #   否则 HTTP/1.1 状态机直接抛 CannotSendRequest("Idle")。
        self._http_busy = False
        # 发命令专用连接。长轮询那条常年被占着，
        # 发命令不能去挤它 —— 各走各的。
        self._http_send = None
        self._send_lock = threading.RLock()

    # ------------------------------------------------------------------
    # 对外
    # ------------------------------------------------------------------

    def start(self, url: str) -> bool:
        """开始连中转。

        url 支持用**分号**分隔多个地址，按顺序尝试：
            http://192.168.1.10:8080;https://aaa.bbb.com

        为什么需要多个：
          * 电脑和中转常在同一局域网 —— 直连局域网中转更快更稳，
            而且地址永远不变（不用管隧道地址变来变去）
          * 电脑在外面时才走隧道

        注意：**地址由用户自己填**，代码里不内置任何默认中转地址。
        """
        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

    @property
    def running(self) -> bool:
        """中转链路现在是不是真的在跑。

        对外给 server.enable_relay() 判断用 ——
        那边以前靠"对象建过没有"来判成功，会在没配地址时谎报。
        """
        return bool(self._running)

    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]
            # dedicated=True：走发命令专用连接，不和长轮询挤一条。
            # 这是「按一下没反应」的修复点 —— 以前这里和长轮询共用
            # 一条 HTTP/1.1 连接，撞上状态机就抛 "Idle"，
            # 异常一路冒到 client-relay 线程外面把它打死。
            self._post("/v1/agent/send", piece, {"X-Token": self.token},
                       timeout=POLL_TIMEOUT, dedicated=True)

    # ------------------------------------------------------------------
    # 内部
    # ------------------------------------------------------------------

    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._fail_streak = getattr(self, "_fail_streak", 0) + 1
                if self._fail_streak >= OFFLINE_AFTER:
                    self.online = False
                self.last_error = f"{type(exc).__name__}: {exc}"
                self._log(f"中转连接出问题（第 {self._fail_streak} 次）："
                          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._fail_streak = 0
        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, dedicated: bool = False) -> None:
        """把缓存的连接关掉（出错后调用，下次会重开）。

        dedicated=True 时清的是"发命令专用连接"，
        否则清长轮询用的那条。
        """
        if dedicated:
            with self._send_lock:
                connection, self._http_send = self._http_send, None
            if connection is not None:
                try:
                    connection.close()
                except Exception:                         # noqa: BLE001
                    pass
            return
        with self._http_lock:
            connection, self._http = self._http, None
            self._http_busy = False
        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,
              dedicated: 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):
            # 取一条可以独占使用的连接：
            #   * dedicated=True  -> 发命令专用连接（自己一把锁锁全程）
            #   * 普通请求        -> 缓存连接；**如果它正忙（长轮询在途）
            #                        就临时开一条，用完即关**
            # 关键：锁必须盖住整个"发请求 + 读响应"周期，
            # 不能只锁"取连接"那一下 —— 否则两个线程会用同一条连接。
            if dedicated:
                self._send_lock.acquire()
                connection = self._http_send
                if connection is None:
                    connection = self._open_http()
                    self._http_send = connection
                temporary = False
            else:
                with self._http_lock:
                    if self._http is None:
                        self._http = self._open_http()
                    if self._http_busy:
                        # 缓存连接正被长轮询占着，临时另开一条
                        connection = self._open_http()
                        temporary = True
                    else:
                        connection = self._http
                        self._http_busy = True
                        temporary = False
            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(dedicated=dedicated)
                if attempt == 2:
                    break
                self._log_quiet(f"连接断了，重开一次再试（{exc}）")
            finally:
                # 无论成功失败都要把"占用"还回去，
                # 否则这条连接会被永久标记为忙，之后每次请求都另开临时连接。
                if dedicated:
                    self._send_lock.release()
                elif not temporary:
                    with self._http_lock:
                        self._http_busy = False
                else:
                    try:
                        connection.close()
                    except Exception:                     # noqa: BLE001
                        pass

        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"