#!/usr/bin/env python3 # -*- coding: utf-8 -*- """局域网遥控 · 中转服务器 作用:让**不在同一网络**的被控端和手表连上。 手表 ──HTTPS──┐ ├──> Cloudflare 边缘 ──隧道──> cloudflared ──> 本程序 :8080 被控端 ─HTTPS─┘ 设计要点 -------- 1. **盲转发**:本程序不认识上层协议,只按「配对码」把两条连接的字节流对上。 配对、令牌校验、权限判断全部由被控端自己做 —— 所以就算这台中转机 被攻破,也进不了用户的电脑。 2. **长轮询**(不是 WebSocket):HTTP 长轮询在任何代理/4G 网络下都最稳, 而且两端用标准库就能实现(被控端是 Python 标准库,手表是纯 Java 无依赖)。 3. **省内存**(这台机器只有 460MB): * 队列用 deque(maxlen) 封顶,客户端不取就丢最旧的,绝不无限增长 * 单条消息和整队都有字节上限 * 线程按需创建,空闲就退出 * 不引任何第三方库 接口 ---- 被控端: POST /v1/agent/register {"deviceId","code","name","version"} -> {"ok","token","code"} POST /v1/agent/poll 头 X-Token -> 原始字节(长轮询,最多 25 秒) POST /v1/agent/send 头 X-Token,体=原始字节 -> {"ok"} POST /v1/agent/bye 头 X-Token -> {"ok"} 控制端: POST /v1/client/attach {"code","name"} -> {"ok","session","deviceName","online"} POST /v1/client/poll 头 X-Session -> 原始字节 POST /v1/client/send 头 X-Session,体=原始字节 -> {"ok"} POST /v1/client/detach 头 X-Session -> {"ok"} 公共: GET /v1/health -> {"ok","devices","sessions",...} GET / 管理页(要 key) """ from __future__ import annotations import json import os import re import secrets import socket import sys import threading import time import urllib.parse from urllib.parse import quote import urllib.request from collections import deque from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path VERSION = "2.1.0" # ---------------- 参数(都能用环境变量覆盖)---------------- PORT = int(os.environ.get("RELAY_PORT", "8080")) BIND = os.environ.get("RELAY_BIND", "127.0.0.1") DATA_DIR = Path(os.environ.get("RELAY_DATA", "/root/lanremote-relay")) # 长轮询最多挂多久。太长会占着 Cloudflare 的连接,太短则空转请求多。 # 长轮询挂多久。 # 25 -> 60:隧道每次请求要多花 1~2 秒(CF 边缘路由 + 回源), # 而这个开销是按"次数"收的。60 秒一循环把请求数降到 1/4, # 暴露在丢包/重连窗口里的概率也降到 1/4。 # **不会让命令变慢**:有数据时立刻返回(靠 device.event 唤醒), # 拉长只是减少空转重开。 # Cloudflare 单请求上限 100 秒,60 留了 40 秒余量(实测 4/4 通过)。 POLL_SECONDS = float(os.environ.get("RELAY_POLL", "60")) # 单个设备最多缓多少字节没发出去。手表半天不取就丢最旧的,保住内存。 MAX_QUEUE_BYTES = int(os.environ.get("RELAY_QUEUE_BYTES", str(2 * 1024 * 1024))) # 单次请求体上限(一个分块 256KB + JSON 外壳,给到 4MB 很宽松) MAX_BODY = int(os.environ.get("RELAY_MAX_BODY", str(4 * 1024 * 1024))) # 配对码:6 位数字(和手表上输的一致) CODE_RE = re.compile(r"^\d{6}$") # 设备多久没心跳算掉线 DEVICE_TIMEOUT = 90.0 # 设备多久没动静就**从列表里彻底删掉**。 # # 和 DEVICE_TIMEOUT(90 秒)不一样:那个只是标记"离线", # 记录还留着,方便手表看到"我以前连过这台"。 # 但记录只增不减会出问题 —— 被控端每次换端口/重装都用新 deviceId # 注册一次,老的永远留着。实测一天攒了 75 条僵尸记录, # /v1/health 从几 KB 涨到 20KB,手表和网页每次白下载一大坨。 # 6 小时足够覆盖"上课中途断一下",真回来会重新注册。 DEVICE_TTL = float(os.environ.get("RELAY_DEVICE_TTL", str(6 * 3600))) # 一个设备最多同时挂几个手表会话 MAX_SESSIONS_PER_DEVICE = 3 # 防爆破:同一个 IP 每小时最多试几次配对码 ATTACH_LIMIT = int(os.environ.get("RELAY_ATTACH_LIMIT", "30")) ATTACH_WINDOW = 3600.0 ADMIN_KEY = os.environ.get("RELAY_ADMIN_KEY", "") def admin_key() -> str: """管理页 / 管理接口的钥匙。 没配置就自动生成一个并记到 data/admin_key.txt。 以前默认是空字符串,等于管理页谁都能打开 —— 那上面能看到配对码,太不合适了。 """ global ADMIN_KEY if ADMIN_KEY: return ADMIN_KEY path = DATA_DIR / "admin_key.txt" try: if path.is_file(): ADMIN_KEY = path.read_text(encoding="utf-8").strip() if not ADMIN_KEY: ADMIN_KEY = secrets.token_urlsafe(18) DATA_DIR.mkdir(parents=True, exist_ok=True) path.write_text(ADMIN_KEY, encoding="utf-8") try: os.chmod(path, 0o600) except OSError: pass except Exception: # noqa: BLE001 ADMIN_KEY = ADMIN_KEY or secrets.token_urlsafe(18) return ADMIN_KEY # 管理页密码。 # # ⚠ 这里**不放默认密码**(以前放了一个写死的,是个严重问题): # 代码一旦进公开仓库,那个密码就等于公开了 —— # 而管理页路径和公网地址也是公开的,等于把管理台的门钥匙挂在门上。 # 就算是私人自用,也不该把密码提交进 git:git 历史删不掉。 # # 正确做法和 admin_key 一样:首次启动**随机生成**,写进 # data/admin_password.txt(权限 600),只有本机 root 能看。 # 想指定就用环境变量 RELAY_ADMIN_PASSWORD。 ADMIN_PASSWORD_DEFAULT = "" # 管理页账号。账号本身不是秘密(还得配密码),但也允许覆盖。 ADMIN_USER_DEFAULT = "xmz7891" ADMIN_USER = (os.environ.get("RELAY_ADMIN_USER") or ADMIN_USER_DEFAULT) ADMIN_PASSWORD = os.environ.get("RELAY_ADMIN_PASSWORD", "") # 登录会话:cookie 令牌 -> 过期时间 # 成品下载目录(管理页上给人下载 APK / 整包用)。 # 只放白名单后缀的文件,文件名做严格校验防目录穿越。 DOWNLOADS_DIR = Path(__file__).resolve().parent / "downloads" # 抢救报告目录。rescue.sh 抢救失败时往这里写 .txt 诊断报告。 # 和 downloads/ 分开:那个是给用户的安装包(带长缓存), # 这个是给管理员看的诊断(必须不缓存)。 RESCUE_DIR = Path(__file__).resolve().parent / "rescue_reports" RESCUE_KEEP = 30 # 只保留最近 30 份(460MB 的小机器,别塞满) def _safe_rescue_name(name: str) -> str: """校验抢救报告文件名。只允许 <时间戳>.txt 这种形状。""" name = (name or "").strip() if not name or len(name) > 80: return "" if "/" in name or "\\" in name or ".." in name: return "" if not name.endswith(".txt"): return "" for ch in name: if ch.isalnum() or ch in "._-": continue return "" return name def _list_rescue_reports() -> list: """列出抢救报告(按时间倒序)。 顺带读每个文件的前几行,抽出结论 —— 后台列表上直接显示, 不用一个个点进去看。 """ items = [] try: if not RESCUE_DIR.is_dir(): return items for entry in sorted(RESCUE_DIR.glob("*.txt"), reverse=True): if not entry.is_file(): continue safe = _safe_rescue_name(entry.name) if not safe: continue stat = entry.stat() summary = "" try: with entry.open("r", encoding="utf-8", errors="replace") as fh: head = fh.read(1200) for line in head.splitlines(): s = line.strip() if s.startswith("判断:"): summary = s.replace("判断:", "").strip() break if not summary: for line in head.splitlines(): s = line.strip() if s.startswith("公网 ") and "->" in s: summary = s break except OSError: pass items.append({ "name": entry.name, "size": stat.st_size, "mtime": stat.st_mtime, "summary": summary[:120], }) except OSError: return items items.sort(key=lambda i: i["mtime"], reverse=True) return items[:RESCUE_KEEP] DOWNLOAD_SUFFIXES = (".apk", ".zip", ".txt", ".exe", ".idsig") def _safe_download_name(name: str) -> str: """校验下载文件名。不合法返回空串。 防目录穿越:只允许字母数字、点、横线、下划线、中文, 且必须是白名单后缀。 """ name = (name or "").strip() if not name or len(name) > 120: return "" if "/" in name or "\\" in name or ".." in name: return "" if not name.lower().endswith(DOWNLOAD_SUFFIXES): return "" for ch in name: if ch.isalnum() or ch in "._-()() ": continue return "" return name def _list_downloads() -> list: """列出可下载的文件(按修改时间倒序)。""" items = [] try: if not DOWNLOADS_DIR.is_dir(): return items for entry in DOWNLOADS_DIR.iterdir(): if not entry.is_file(): continue if not _safe_download_name(entry.name): continue stat = entry.stat() items.append({ "name": entry.name, "size": stat.st_size, "mtime": stat.st_mtime, }) except OSError: return items items.sort(key=lambda item: item["mtime"], reverse=True) return items ADMIN_SESSIONS: dict = {} ADMIN_SESSION_SECONDS = 12 * 3600 # 管理后台的路径。 # 用户要求官网放 /(公开),管理后台另开一个不那么好猜的路径。 # # 之前这个常量其实**从来没定义过** —— 路由里写的是字面量 # "/zhongzhuanzhan",而新写的 _admin_page 引用了 ADMIN_PATH, # 结果一访问管理页就 NameError(页面 000)。 # 现在统一定义,路由和页面都引用它,避免再次不一致。 ADMIN_PATH = "/zhongzhuanzhan" def admin_password() -> str: """管理页密码。优先级:环境变量 -> data/admin_password.txt -> 随机生成。 ⚠ 为什么是"随机生成"而不是"写死一个默认值": 写死的默认值一旦进了公开仓库,就等于把管理台钥匙公开 —— 而且 git 历史删不掉,改代码也补不回来。 随机生成 + 写进 600 权限的文件,只有本机能看。 """ global ADMIN_PASSWORD if ADMIN_PASSWORD: return ADMIN_PASSWORD path = DATA_DIR / "admin_password.txt" try: if path.is_file(): ADMIN_PASSWORD = path.read_text(encoding="utf-8").strip() if not ADMIN_PASSWORD: if ADMIN_PASSWORD_DEFAULT: # 兼容:确实想指定一个固定值(不推荐,别提交进仓库) ADMIN_PASSWORD = ADMIN_PASSWORD_DEFAULT else: # 生成一个好念但够强的密码:4 段,每段 4 个字符 # 用 token_urlsafe 保证随机性,去掉易混字符 raw = secrets.token_urlsafe(12) ADMIN_PASSWORD = "".join( ch for ch in raw if ch.isalnum())[:16] DATA_DIR.mkdir(parents=True, exist_ok=True) path.write_text(ADMIN_PASSWORD, encoding="utf-8") try: os.chmod(path, 0o600) except OSError: pass log("=" * 56) log(" 已生成管理页密码(只显示这一次,也写在下面这个文件里)") log(" 账号:%s" % ADMIN_USER) log(" 密码:%s" % ADMIN_PASSWORD) log(" 文件:%s" % path) log(" 想改就直接改那个文件,或者设 RELAY_ADMIN_PASSWORD。") log(" ⚠ 不要把密码提交进 git —— 历史删不掉。") log("=" * 56) except Exception: # noqa: BLE001 # 落盘失败也不能让管理页打不开:退回一个进程内随机密码 ADMIN_PASSWORD = ADMIN_PASSWORD or "".join( ch for ch in secrets.token_urlsafe(12) if ch.isalnum())[:16] return ADMIN_PASSWORD def admin_login(username: str, password: str) -> str: """验密码,对了就发一个会话令牌。""" # 账号也要对(用户要求用账号+密码登录) if username and username.strip() != ADMIN_USER: return "" if not password or not secrets.compare_digest(password, admin_password()): return "" token = secrets.token_urlsafe(24) now = time.time() # 顺手清掉过期的,别让字典涨 for old, expiry in list(ADMIN_SESSIONS.items()): if expiry < now: ADMIN_SESSIONS.pop(old, None) ADMIN_SESSIONS[token] = now + ADMIN_SESSION_SECONDS return token def admin_session_valid(token: str) -> bool: if not token: return False expiry = ADMIN_SESSIONS.get(token) if expiry is None: return False if expiry < time.time(): ADMIN_SESSIONS.pop(token, None) return False return True def admin_ok(supplied: str) -> bool: """命令行/脚本用的通道:钥匙或者密码,常数时间比较。""" if not supplied: return False if secrets.compare_digest(supplied, admin_key()): return True return secrets.compare_digest(supplied, admin_password()) def public_url() -> str: """当前隧道的公网地址。 隧道换地址时会写进 public_url.txt(tunnel-run.sh 负责维护)。 这里几个可能的位置都找一下,找到第一个像地址的就返回。 """ candidates = [ DATA_DIR.parent / "public_url.txt", DATA_DIR / "public_url.txt", Path("/root/lanremote-relay/public_url.txt"), ] for path in candidates: try: if not path.is_file(): continue text = path.read_text(encoding="utf-8").strip() if text.startswith("http"): return text except OSError: continue return "" def log(message: str) -> None: stamp = time.strftime("%m-%d %H:%M:%S") print(f"[{stamp}] {message}", flush=True) # ====================================================================== # 数据 # ====================================================================== class Device: """一台被控端。""" __slots__ = ("device_id", "code", "name", "version", "token", "last_seen", "connected_at", "to_client", "to_agent", "event", "bytes_in", "bytes_out", "current_session", "active_session", "lan_ips", "port") def __init__(self, device_id: str, code: str, name: str, version: str = "") -> None: self.device_id = device_id self.code = code self.name = name self.version = version self.token = secrets.token_hex(24) # 这台电脑的局域网 IP 和监听端口。 # 手表拿它判断"要不要切成局域网直连"。 self.lan_ips: list = [] self.port: int = 0 self.last_seen = time.time() self.connected_at = time.time() self.to_client = bytearray() # 被控端 -> 手表 self.to_agent = bytearray() # 手表 -> 被控端 self.event = threading.Event() # 有新数据时唤醒长轮询 self.bytes_in = 0 self.bytes_out = 0 # 当前挂在这台设备上的手表会话令牌。变了就说明手表换了一台/重连了, # 被控端据此重开一条会话,避免把上一条的状态带过来。 self.current_session = "" # 当前挂在这台设备上的 Session 对象。数据来了只唤醒它, # 不要让已经被顶掉的旧会话把数据截胡。 self.active_session = None def online(self) -> bool: return (time.time() - self.last_seen) < DEVICE_TIMEOUT def push_to_agent(self, data: bytes) -> None: if not data: return self.bytes_in += len(data) self._append(self.to_agent, data) self.event.set() def push_to_client(self, data: bytes) -> None: if not data: return self.bytes_out += len(data) self._append(self.to_client, data) self.event.set() # 同时叫醒当前那个会话的长轮询(旧会话已经被替换,不该再拿数据) session = self.active_session if session is not None: session.event.set() @staticmethod def _append(buffer: bytearray, data: bytes) -> None: """往队列里塞数据,超上限就丢最旧的 —— 内存优先。""" buffer.extend(data) overflow = len(buffer) - MAX_QUEUE_BYTES if overflow > 0: del buffer[:overflow] class Session: """一个手表连接。""" __slots__ = ("token", "device_id", "code", "name", "ip", "created", "last_seen", "event", "replaced") def __init__(self, device_id: str, code: str, name: str, ip: str) -> None: self.token = secrets.token_hex(24) self.device_id = device_id self.code = code self.name = name self.ip = ip self.created = time.time() self.last_seen = time.time() # 自己的唤醒事件。每个会话独立等待,避免被顶掉的旧会话截胡数据。 self.event = threading.Event() self.replaced = False class SessionReplaced(Exception): """这个会话已经被新的手表顶替了 —— 它必须让位,不能继续取数据。""" class Relay: def __init__(self) -> None: self.lock = threading.RLock() self.devices: dict = {} # device_id -> Device self.codes: dict = {} # code -> device_id self.sessions: dict = {} # session_token -> Session self.attach_log: dict = {} # ip -> [时间戳] # 被新连接顶替掉的会话:token -> 原因。 # 留着是为了能明确告诉旧客户端"你是被顶掉的", # 而不是含糊的 401 —— 否则它会以为自己只是过期了,然后抢回来。 self.retired: dict = {} # 是否允许新设备接入。False 时只有已登记过的 deviceId 能上线。 self.allow_new = True # 设备历史名单(只有 deviceId,不含配对码)。 # **超时清理绝不动它** —— 它是"这台设备曾经被允许过"的唯一凭证。 # 详见 register() 里 known_device 的注释。 self.known_devices: set = set() self.started = time.time() self.total_requests = 0 self.state_file = DATA_DIR / "relay_state.json" self._load() # ---------------- 持久化(只存配对码,恢复单设备单配对码制)-------- def _load(self) -> None: try: raw = json.loads(self.state_file.read_text(encoding="utf-8")) except Exception: # noqa: BLE001 return # 是否允许新设备接入(重启不丢) self.allow_new = bool(raw.get("allowNew", True)) # 恢复设备历史名单(重启不丢) for did in raw.get("knownDevices", []): did = str(did or "").strip() if did: self.known_devices.add(did) for item in raw.get("codes", []): device_id = str(item.get("deviceId", "")) code = str(item.get("code", "")) if device_id and CODE_RE.match(code): self.codes[code] = device_id log(f"恢复配对码绑定:设备 {device_id[:12]}… -> {code}") def _save(self) -> None: try: DATA_DIR.mkdir(parents=True, exist_ok=True) payload = { "allowNew": bool(self.allow_new), # 设备历史:超时清理不会删它,所以关掉接入开关后 # 老设备第二天开机仍然能上线 "knownDevices": sorted(self.known_devices), "codes": [ {"deviceId": did, "code": code} for code, did in sorted(self.codes.items()) ], } tmp = self.state_file.with_suffix(".tmp") tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") tmp.replace(self.state_file) except Exception as exc: # noqa: BLE001 log(f"保存状态失败:{exc}") # ---------------- 被控端 ---------------- def register(self, body: dict) -> dict: """被控端上线登记。 **单设备单配对码制**: * 一台设备第一次来 -> 记下它报的码 * 同一台设备(deviceId 一样)以后再来 -> 一律用它**原来那个码**, 它要是报了别的码,告诉它用旧的 * 一个码不能同时属于两台设备 -> 拒绝 """ device_id = str(body.get("deviceId") or "").strip() code = str(body.get("code") or "").strip() name = str(body.get("name") or "").strip()[:40] or "未命名设备" version = str(body.get("version") or "").strip()[:20] # 这台电脑的局域网 IP + 端口(手表判断要不要走局域网直连) raw_ips = body.get("lanIps") lan_ips = [] if isinstance(raw_ips, list): for item in raw_ips[:8]: text = str(item).strip() # 只收看起来像 IPv4 的,别把乱七八糟的东西转给手表 parts = text.split(".") if len(parts) == 4 and all( p.isdigit() and 0 <= int(p) <= 255 for p in parts): lan_ips.append(text) try: listen_port = int(body.get("port") or 0) except (TypeError, ValueError): listen_port = 0 if not (0 < listen_port < 65536): listen_port = 0 if not device_id or len(device_id) > 80: return {"ok": False, "error": "deviceId 不合法"} # deviceId 走白名单:字母数字 + - _ . : # # 为什么要卡这么死: # * 控制字符(\x00、\r\n)能进来 —— 管理页转义做得对所以 # 不是 XSS,但 deviceId 会写进日志,`\r\n` 能伪造日志行 # * 路径分隔符(/ \)能进来 —— 会进状态文件和页面,不该有 # # 真实 deviceId 的形状是「机器名-12位hex」(见 _ensure_device_id), # 字符集本来就很窄,卡严不会有任何副作用。 if not all(ch.isalnum() or ch in "-_.:" for ch in device_id): log(f"deviceId 含非法字符,拒绝:{device_id!r}") return {"ok": False, "error": "deviceId 含非法字符"} if not CODE_RE.match(code): return {"ok": False, "error": "配对码必须是 6 位数字"} with self.lock: # 管理员关掉了"允许新设备接入"时, # 只放行**已经登记过**的设备。 # # 注意判断依据是 deviceId(装机时生成、重装也不变), # **不是配对码** —— 否则管理员给老设备改了码, # 那台设备就被自己的开关挡在外面了。 # ⚠ 判断"老设备"必须查 known_devices(设备历史), # **不能查 self.devices** —— devices 里超过 6 小时没上报的 # 会被 snapshot() 清掉(电脑关机过夜必然超时), # 那时老设备会被自己的开关拒之门外,而且管理员在页面上 # 已经看不到它、不知道该放开谁,重启也救不回来。 known_device = device_id in self.known_devices if not known_device and not self.allow_new: log(f"新设备接入已被关闭,拒绝 {name} ({device_id[:12]}…)") return {"ok": False, "blocked": True, "error": "中转站当前不接受新设备接入"} known = self.codes.get(code) if known and known != device_id: # codeTaken 让客户端能精确识别这种冲突, # 不用去匹配中文错误文案(文案一改就坏)。 log(f"配对码 {code} 已被 {known} 占用,拒绝 {device_id}") return {"ok": False, "codeTaken": True, "error": "该配对码已被使用,请换一个配对码"} # 这台设备以前配过码吗? old_code = None for existing_code, existing_device in self.codes.items(): if existing_device == device_id: old_code = existing_code break if old_code and old_code != code: log(f"设备 {name} 报了新码 {code},按单设备单码制改回 {old_code}") code = old_code self.codes[code] = device_id device = self.devices.get(device_id) if device is None: device = Device(device_id, code, name, version) self.devices[device_id] = device else: device.code = code device.name = name device.version = version device.token = secrets.token_hex(24) device.last_seen = time.time() # 记进设备历史名单(这是"曾经被允许过"的凭证,只增不减; # 只有管理员显式删除设备时才会被移除) self.known_devices.add(device_id) # 每次登记都刷新局域网信息(IP 可能变,端口也可能改) device.lan_ips = lan_ips device.port = listen_port self._save() log(f"被控端上线:{name} 码 {code} ({device_id[:12]}…)") note = "" if old_code and old_code != str(body.get("code") or "").strip(): note = (f"这台设备之前设过配对码 {old_code}," f"已按「单设备单配对码」沿用它") return {"ok": True, "token": device.token, "code": code, "note": note, "pollSeconds": POLL_SECONDS} # ---------------- 管理操作 ---------------- def set_allow_new(self, allow: bool) -> dict: """开关"允许新设备接入"。""" with self.lock: self.allow_new = bool(allow) self._save() log(f"管理员把「允许新设备接入」设为 {self.allow_new}") return {"ok": True, "allowNew": self.allow_new} def set_device_code(self, device_id: str, new_code: str) -> dict: """给一台设备换配对码。 要点: * 新码必须是 6 位数字 * 不能已经被**别的**设备占用 * 旧码的映射必须删掉,否则会变成孤儿条目, 管理页上"配对码占用"的数字会虚高 * 该设备挂着的会话要踢掉 —— 会话是绑在码上的, 不踢的话手表还拿着旧会话以为连着 """ device_id = str(device_id or "").strip() new_code = str(new_code or "").strip() if not CODE_RE.match(new_code): return {"ok": False, "error": "配对码必须是 6 位数字"} with self.lock: device = self.devices.get(device_id) if device is None: return {"ok": False, "error": "没有这台设备(可能已下线)"} owner = self.codes.get(new_code) if owner and owner != device_id: return {"ok": False, "error": "这个配对码已经被别的设备占用了"} old_code = device.code if old_code == new_code: return {"ok": True, "code": new_code, "unchanged": True} # 删掉旧码映射(连同可能存在的别名) for code, owner_id in list(self.codes.items()): if owner_id == device_id: self.codes.pop(code, None) self.codes[new_code] = device_id device.code = new_code # 会话绑在码上,旧会话作废,让手表重新配对。 # # ⚠ 会话是以 token 为 key 存在 self.sessions 里的, # Session 用了 __slots__,**根本没有 session_id 属性**。 # 必须遍历 items() 拿 key;写 s.session_id 会 AttributeError, # 而且会被上层兜底成"服务器内部错误"(500),很难定位。 kicked = 0 for sid in [t for t, s in self.sessions.items() if s.device_id == device_id]: self.sessions.pop(sid, None) kicked += 1 self._save() log(f"管理员把设备 {device.name} 的配对码 " f"{old_code} 改成 {new_code}(踢掉 {kicked} 个会话)") return {"ok": True, "code": new_code, "oldCode": old_code, "kicked": kicked} def device_by_token(self, token: str) -> Device | None: if not token: return None with self.lock: for device in self.devices.values(): if secrets.compare_digest(device.token, token): return device return None # ---------------- 控制端 ---------------- def attach(self, body: dict, ip: str) -> dict: code = str(body.get("code") or "").strip() name = str(body.get("name") or "").strip()[:40] or "手表" if not CODE_RE.match(code): return {"ok": False, "error": "配对码要是 6 位数字"} with self.lock: now = time.time() hits = [t for t in self.attach_log.get(ip, []) if now - t < ATTACH_WINDOW] if len(hits) >= ATTACH_LIMIT: wait = int(ATTACH_WINDOW - (now - hits[0])) log(f"配对尝试过多:{ip}") return {"ok": False, "error": f"尝试次数过多,请等 {wait} 秒"} hits.append(now) self.attach_log[ip] = hits device_id = self.codes.get(code) if not device_id: return {"ok": False, "error": "没有设备在用这个配对码", "needPair": True} device = self.devices.get(device_id) if device is None or not device.online(): return {"ok": False, "error": "设备不在线", "offline": True, "deviceName": device.name if device else ""} # 一台设备同时只留一个手表会话。 # (之前留 3 个,但被控端只有一条到中转的流,分不清哪段字节 # 属于哪个手表 —— 所以这里改成新的来了就踢旧的。) for stale_token, stale in list(self.sessions.items()): if stale.device_id == device_id: self.sessions.pop(stale_token, None) self.retired[stale_token] = "replaced" stale.replaced = True # 立刻叫醒它挂着的长轮询 —— 否则它会一直等下去, # 等新数据来时把本该给新手表的数据抢走。 stale.event.set() log(f"踢掉旧会话 {stale_token[:8]}…(同一设备只允许一个)") # 清掉太老的顶替记录,别让字典涨 if len(self.retired) > 200: for old_token in list(self.retired)[:100]: self.retired.pop(old_token, None) session = Session(device_id, code, name, ip) self.sessions[session.token] = session device.current_session = session.token device.active_session = session # 换会话了:把上一条留在队列里的半截数据清掉, # 不然新手表会收到上一个手表没取走的残留字节。 device.to_client.clear() log(f"手表接入:{name} ({ip}) -> {device.name}") return {"ok": True, "session": session.token, "deviceName": device.name, "online": True, # 手表拿这两个判断"是不是和电脑在同一个网络", # 是的话就切成局域网直连(快得多)。 "lanIps": list(device.lan_ips), "port": device.port, "pollSeconds": POLL_SECONDS} def session_by_token(self, token: str) -> Session | None: if not token: return None with self.lock: session = self.sessions.get(token) if session is None: return None session.last_seen = time.time() return session def session_replaced(self, token: str) -> bool: """这个令牌是不是「被新连接顶掉」的(而不是过期/不存在)。""" if not token: return False with self.lock: return self.retired.get(token) == "replaced" # ---------------- 收发 ---------------- def agent_take(self, device: Device, wait: float) -> bytes: """被控端取走「手表发来的」数据;没有就挂着等。""" return self._take(device, device.to_agent, wait, "agent") def agent_put(self, device: Device, data: bytes) -> None: """被控端把「要发给手表」的数据放进来。""" device.push_to_client(data) def client_take(self, session: Session, wait: float) -> bytes: """手表取走「被控端发来的」数据。 注意这里**不能**复用 _take:那个是按设备的事件等的, 会把被顶掉的旧会话和新会话混在一起等,导致数据被旧会话截胡。 每个会话等自己的 event,并且每轮先看自己是不是已经被顶替了。 """ deadline = time.time() + wait while True: if session.replaced: raise SessionReplaced() with self.lock: device = self.devices.get(session.device_id) if device is None: return b"" if device.to_client: data = bytes(device.to_client) device.to_client.clear() device.event.clear() session.event.clear() return data remaining = deadline - time.time() if remaining <= 0: return b"" session.event.wait(min(remaining, 0.5)) session.event.clear() def client_put(self, session: Session, data: bytes) -> None: """手表把「要发给被控端」的数据放进来。""" with self.lock: device = self.devices.get(session.device_id) if device is not None: device.push_to_agent(data) def _take(self, device: Device, buffer: bytearray, wait: float, who: str) -> bytes: """从队列里取数据。空了就等,最多 wait 秒。 用 device.event 做唤醒:另一端 push 时会 set(),这里立刻醒来, 不用轮询 —— 省 CPU,这台机器只有 4 个核还要跑 4G 模块。 """ deadline = time.time() + wait while True: if who == "agent": device.last_seen = time.time() with self.lock: if buffer: data = bytes(buffer) buffer.clear() device.event.clear() return data remaining = deadline - time.time() if remaining <= 0: if who == "agent": device.last_seen = time.time() return b"" device.event.wait(min(remaining, 1.0)) device.event.clear() # ---------------- 统计 / 管理 ---------------- def snapshot(self) -> dict: with self.lock: now = time.time() # 先把长时间没动静的设备清掉。 # # 为什么必须清:这个 dict 只增不减。被控端每次启动/换端口/ # 重装都会用新的 deviceId 注册一次,老的永远留着 —— # 实测跑一天就攒了 75 条僵尸记录,/v1/health 从几 KB # 涨到 20KB,手表和网页每次都要白下载一大坨没用的东西。 # 设备超过 6 小时没上报就当它下线了(真回来会重新注册)。 stale = [did for did, dev in self.devices.items() if now - getattr(dev, "last_seen", now) > DEVICE_TTL] for did in stale: self.devices.pop(did, None) # 它的会话也一起收掉,免得挂着一堆死会话。 # ⚠ 同样要用 items() 拿 token 当 key —— # Session 没有 session_id 属性(见 set_device_code 里的说明)。 # 这段以前写的是 s.session_id,只是因为平时 stale 是空的 # 才一直没炸;一旦真有设备超时就会 500。 for sid in [t for t, s in self.sessions.items() if s.device_id == did]: self.sessions.pop(sid, None) # 配对码映射也要一起清。 # 它和 devices 是一对孪生 dict,只清一个的话, # codes 里会攒下一堆指向已删设备的孤儿条目 —— # 实测管理页显示"配对码占用 131",实际一台设备都没有。 # 指向已不存在设备的码 = 废码,直接扔。 orphan = [code for code, owner in self.codes.items() if owner not in self.devices] for code in orphan: self.codes.pop(code, None) # ⚠ 清完孤儿码要落盘。不然内存里没了、磁盘上还在, # 重启后 _load() 又恢复回来,下一次 snapshot() 再清一遍 —— # 看上去能用,但每次重启都会"回滚"一次状态,很难排查。 # (超时清理 devices 也有同样问题,一并落盘。) if stale or orphan: self._save() devices = [] for device in self.devices.values(): sessions = [s for s in self.sessions.values() if s.device_id == device.device_id] devices.append({ "deviceId": device.device_id, "code": device.code, "name": device.name, # 手表用这两个判断能不能走局域网直连 "lanIps": list(device.lan_ips), "port": device.port, "version": device.version, "online": device.online(), "idleSeconds": int(now - device.last_seen), "sessions": len(sessions), "queuedToClient": len(device.to_client), "queuedToAgent": len(device.to_agent), "bytesIn": device.bytes_in, "bytesOut": device.bytes_out, }) devices.sort(key=lambda d: d["name"]) return { "ok": True, "version": VERSION, "uptimeSeconds": int(now - self.started), "devices": devices, "sessions": len(self.sessions), "totalRequests": self.total_requests, "memoryKB": _rss_kb(), # 管理页上的"允许新设备接入"开关就显示这个值 "allowNew": bool(self.allow_new), "codeCount": len(self.codes), } def kick(self, device_id: str) -> int: with self.lock: gone = [t for t, s in self.sessions.items() if s.device_id == device_id] for token in gone: self.sessions.pop(token, None) device = self.devices.get(device_id) if device is not None: device.to_agent.clear() device.to_client.clear() device.last_seen = 0 # 标成掉线,逼它重连 return len(gone) def delete_device(self, *, device_id: str = "", code: str = "") -> dict: """删掉一台设备,**连同它的配对码绑定和所有会话**。 这就是管理页上「删除设备」按钮做的事。 和 kick/forget 的区别: kick 只踢会话,设备还在,码还占着 forget 只抹码绑定,设备还挂在列表上 delete 三样一起清 —— 那台设备下次可以重新设任意配对码 返回做了什么,便于界面显示。 """ target = None with self.lock: if device_id: target = self.devices.get(device_id) if target is None and code: owner = self.codes.get(code) if owner: target = self.devices.get(owner) device_id = owner if target is None: return {"ok": False, "error": "找不到这台设备"} removed_sessions = self.kick(target.device_id) with self.lock: self.devices.pop(target.device_id, None) # "删除设备"是**真正忘记这台设备** —— 连历史名单一起删。 # 不删的话,管理员点了删除、开关又关着, # 那台设备下次来仍然被当成老设备放行,开关就形同虚设。 self.known_devices.discard(target.device_id) # 把它的码绑定一起删掉 gone_codes = [c for c, owner in list(self.codes.items()) if owner == target.device_id] for item in gone_codes: self.codes.pop(item, None) # 清掉可能残留的会话记录 for token in list(getattr(self, "sessions", {}).keys()): self.sessions.pop(token, None) self._save() log(f"管理操作:删除设备 {target.name}(码 {target.code})," f"踢掉 {removed_sessions} 个会话,清掉 {len(gone_codes)} 条码绑定") return { "ok": True, "deleted": target.device_id, "name": target.name, "code": target.code, "kicked": removed_sessions, "codesCleared": gone_codes, } def forget(self, code: str) -> bool: """抹掉一个配对码绑定(让那台设备下次可以设新码)。 注意:**只抹码,不抹设备历史**。 抹码的语义是"这个码我不占了",不是"我不认识这台设备了"。 """ with self.lock: device_id = self.codes.pop(code, None) if not device_id: return False device = self.devices.get(device_id) if device is not None: self.devices.pop(device_id, None) self._save() return True def _rss_kb() -> int: """自己占了多少内存(KB)。省内存的项目得能看见自己吃多少。""" try: with open("/proc/self/status", encoding="utf-8") as handle: for line in handle: if line.startswith("VmRSS:"): return int(line.split()[1]) except Exception: # noqa: BLE001 pass return 0 RELAY = Relay() # ====================================================================== # HTTP 层 # ====================================================================== def _query(url: str, name: str) -> str: """从 URL 里取一个查询参数(不引 urllib,省内存)。""" if "?" not in url: return "" try: from urllib.parse import parse_qs values = parse_qs(url.split("?", 1)[1]).get(name) or [] return values[0] if values else "" except Exception: # noqa: BLE001 return "" def _site_page() -> str: """官网首页(公开,不用登录)。 实际渲染在 relay/site.py —— relay.py 已经两千多行, 再放几百行 HTML 就没法看了。网站属于展示层, 跟中转逻辑分开更清楚。 """ # 注意别把模块叫 site.py —— 那是 Python 标准库的名字, # import site 会拿到标准库那个,页面直接崩。 # (这个坑是本地渲染测试抓出来的,部署前就修了。) from landing import render_site return render_site(_list_downloads()) def _login_page(message: str = "") -> bytes: """管理页登录页(输密码,不再是 URL 里带钥匙)。""" notice = ('
' + message + '
') if message else '' html = """