#!/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 = "1.8.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
# 管理页密码。用户指定的默认值，可以在 data/admin_password.txt 里改。
ADMIN_PASSWORD_DEFAULT = "@REVAedu120830"
# 管理页账号（原来只有密码，用户要求加上账号）
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"
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:
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:
ADMIN_PASSWORD = ADMIN_PASSWORD_DEFAULT
DATA_DIR.mkdir(parents=True, exist_ok=True)
path.write_text(ADMIN_PASSWORD, encoding="utf-8")
try:
os.chmod(path, 0o600)
except OSError:
pass
except Exception: # noqa: BLE001
ADMIN_PASSWORD = ADMIN_PASSWORD or ADMIN_PASSWORD_DEFAULT
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 = {}
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
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 = {"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 不合法"}
if not CODE_RE.match(code):
return {"ok": False, "error": "配对码必须是 6 位数字"}
with self.lock:
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()
# 每次登记都刷新局域网信息（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 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)
# 它的会话也一起收掉，免得挂着一堆死会话
for sid in [s.session_id for s in self.sessions.values()
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)
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(),
}
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)
# 把它的码绑定一起删掉
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 = """
中转服务器 · 登录
中转服务器
局域网遥控 · 管理后台
""" + notice + """
"""
return html.encode("utf-8")
class Handler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
server_version = f"LanRemoteRelay/{VERSION}"
# 关掉逐条访问日志 —— 4G 机器上写日志很费
def log_message(self, fmt, *args): # noqa: A003
pass
# ---------------- 小工具 ----------------
def _send(self, code: int, body: bytes = b"",
content_type: str = "application/json",
extra: dict | None = None) -> None:
self.send_response(code)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "no-store")
for name, value in (extra or {}).items():
self.send_header(name, value)
self.end_headers()
if body:
try:
self.wfile.write(body)
except (BrokenPipeError, ConnectionResetError):
pass
def _json(self, code: int, payload: dict) -> None:
self._send(code, json.dumps(payload, ensure_ascii=False).encode("utf-8"))
def _body(self) -> bytes:
"""读请求体。支持 Content-Length 和 chunked 两种。
为什么必须支持 chunked：
Cloudflare 转发时如果拿不到原始 Content-Length，
就会用 Transfer-Encoding: chunked。
而 Python 的 BaseHTTPRequestHandler **不会**自动解码 chunked。
不读走的话，残留的 "0\r\n\r\n" 会被当成下一个请求行，
报 "Bad request syntax ('0')"，整条连接错位 ——
表现就是隧道那边不停 502/530，可用率上不去。
"""
transfer = (self.headers.get("Transfer-Encoding") or "").lower()
if "chunked" in transfer:
return self._body_chunked()
try:
length = int(self.headers.get("Content-Length") or 0)
except ValueError:
return b""
if length <= 0:
return b""
if length > MAX_BODY:
# 超限也要读掉，否则残留数据同样会污染下一条请求
self._drain(length)
return b""
return self._read_exact(length)
def _read_exact(self, length: int) -> bytes:
"""保证读满 length 字节（rfile.read 在慢连接上可能读不满）。"""
chunks = []
remaining = length
while remaining > 0:
piece = self.rfile.read(remaining)
if not piece:
break
chunks.append(piece)
remaining -= len(piece)
return b"".join(chunks)
def _drain(self, length: int) -> None:
"""把多余的请求体读掉丢弃，避免污染下一个请求。"""
remaining = min(length, MAX_BODY * 2)
while remaining > 0:
piece = self.rfile.read(min(remaining, 65536))
if not piece:
break
remaining -= len(piece)
def _body_chunked(self) -> bytes:
"""按 RFC 7230 解码 chunked 请求体。
注意最后那个 0 长度的 chunk 后面还有一段 trailer（通常为空），
一定要读掉 —— 残留就是 "Bad request syntax ('0')" 的来源。
"""
data = bytearray()
while True:
line = self.rfile.readline(256)
if not line:
break
line = line.strip()
if not line:
continue
size_text = line.split(b";")[0].strip()
try:
size = int(size_text, 16)
except ValueError:
break
if size <= 0:
# 末尾 trailer：读到空行为止
while True:
trailer = self.rfile.readline(256)
if not trailer or trailer.strip() == b"":
break
break
if len(data) + size > MAX_BODY:
self._drain(size + 2)
break
data += self._read_exact(size)
self.rfile.readline(8) # 每个 chunk 结尾的 CRLF
return bytes(data)
def _json_body(self) -> dict:
raw = self._body()
if not raw:
return {}
try:
value = json.loads(raw.decode("utf-8", "replace"))
return value if isinstance(value, dict) else {}
except Exception: # noqa: BLE001
return {}
def _redirect(self, location: str) -> None:
self.send_response(302)
self.send_header("Location", location)
self.send_header("Content-Length", "0")
self.end_headers()
def _admin_cookie(self) -> str:
"""从 Cookie 里取管理会话令牌。"""
raw = self.headers.get("Cookie") or ""
for piece in raw.split(";"):
name, _, value = piece.strip().partition("=")
if name == "lr_admin":
return value
return ""
def _admin_gate(self) -> bool:
"""管理页/管理接口的统一入口：Cookie 或 X-Admin 头。"""
if admin_session_valid(self._admin_cookie()):
return True
return admin_ok(self.headers.get("X-Admin") or "")
def _client_ip(self) -> str:
# 走 Cloudflare，真实 IP 在 CF-Connecting-IP 里
return (self.headers.get("CF-Connecting-IP")
or self.headers.get("X-Forwarded-For", "").split(",")[0].strip()
or self.client_address[0])
# ---------------- 路由 ----------------
def do_HEAD(self): # noqa: N802
"""HEAD 请求只回头，不回体。
之前没实现，HEAD 一律 501 Unsupported method。
浏览器用 GET 所以页面照开，但：
* Cloudflare 的健康检查、很多监控/代理探测会用 HEAD
* curl -I 也是 HEAD，排查时看不出响应头很别扭
⚠ 第一版实现有个 bug：它把响应体整个算出来再取长度。
对下载路由（15 MB 的安装包）等于要 read 一遍整个文件 ——
既浪费内存又慢。更糟的是当时**没给 /download/ 写分支**，
于是 HEAD 一个 APK 会返回 `Content-Length: 0` + `text/plain`，
下载器（IDM、迅雷、部分浏览器）据此会认为文件是空的。
现在改成只算**元数据**（状态码 / 类型 / 长度 / 附加头），
一个字节都不读。
"""
RELAY.total_requests += 1
path = self.path.split("?")[0]
try:
info = self._head_info(path)
except Exception: # noqa: BLE001
info = {"status": 500, "ctype": "text/plain; charset=utf-8",
"length": 0, "extra": {}}
self.send_response(info["status"])
self.send_header("Content-Type", info["ctype"])
self.send_header("Content-Length", str(info["length"]))
for name, value in (info.get("extra") or {}).items():
self.send_header(name, value)
# 静态图可以缓存，跟 GET 那边保持一致
if path.startswith("/shot/"):
self.send_header("Cache-Control", "public, max-age=86400")
else:
self.send_header("Cache-Control", "no-store")
self.end_headers()
def _as_bytes(self, value) -> bytes:
"""把页面函数的结果统一转成 bytes。
为什么需要：`_site_page()` 返回 str，`_login_page()` 返回 bytes，
两个同类的函数返回类型不一致。直接对 bytes 调 .encode()
会 AttributeError，而且会被上层兜底悄悄吃掉，
表现为"页面类型莫名变成 text/plain"，很难查。
这个函数把两种情况都兜住。
"""
if isinstance(value, bytes):
return value
if isinstance(value, str):
return value.encode("utf-8")
return b""
def _head_info(self, path: str) -> dict:
"""给 HEAD 用：算这个路径的响应元数据，不读任何大文件。
注意：**不能标 @staticmethod**。
登录页要走 self._login_page()，静态方法里没有 self，
一调用就抛异常，被 do_HEAD 的兜底吃掉，
结果管理页的 HEAD 报告成 text/plain（今天真踩了这次）。
"""
def ok(ctype: str, length: int, extra: dict | None = None) -> dict:
return {"status": 200, "ctype": ctype, "length": length,
"extra": extra or {}}
def no(status: int = 404) -> dict:
return {"status": status, "ctype": "text/plain; charset=utf-8",
"length": 0, "extra": {}}
if path == "/":
body = self._as_bytes(_site_page())
return ok("text/html; charset=utf-8", len(body))
if path in ("/v1/health", "/v1/downloads"):
return ok("application/json", 2)
if path in ("/url", "/v1/url"):
return ok("text/plain; charset=utf-8", len(public_url() or ""))
if path in ("/favicon.ico", "/shot/favicon.ico"):
icon = Path(__file__).resolve().parent / "favicon.ico"
if icon.is_file():
return ok("image/x-icon", icon.stat().st_size)
return no()
if path == ADMIN_PATH or path == "/admin" or path == "/login":
# 注意：_login_page() 返回的**已经是 bytes**（它的签名是 -> bytes），
# 而 _site_page() 返回的是 str（要自己 encode）。
# 两个函数返回值类型不一致，这里踩过一次：
# 对着 bytes 调 .encode() 直接 AttributeError，
# 被 do_HEAD 的兜底吃掉，于是管理页 HEAD 报成 text/plain。
# 用 _as_bytes() 兜住，以后谁改了返回类型都不至于炸。
return ok("text/html; charset=utf-8",
len(self._as_bytes(_login_page())))
# ---- 产品截图 ----
if path.startswith("/shot/"):
name = path[len("/shot/"):]
if ".." in name or "/" in name or "\\" in name:
return no()
target = Path(__file__).resolve().parent / "site_shots" / name
if target.is_file() and name.lower().endswith(
(".png", ".jpg", ".jpeg", ".webp")):
ctype = ("image/png" if name.lower().endswith(".png")
else "image/jpeg")
return ok(ctype, target.stat().st_size)
return no()
# ---- 下载区 ----
# 这一块原来漏了，导致 HEAD 一个 APK 报 Content-Length: 0。
# 类型判断必须和 do_GET 里那份保持一致。
if path.startswith("/download/"):
wanted = urllib.parse.unquote(path[len("/download/"):])
safe = _safe_download_name(wanted)
if not safe:
return no()
target = DOWNLOADS_DIR / safe
if not target.is_file():
return no()
lower = safe.lower()
if lower.endswith(".apk"):
ctype = "application/vnd.android.package-archive"
elif lower.endswith(".zip"):
ctype = "application/zip"
elif lower.endswith(".txt"):
ctype = "text/plain; charset=utf-8"
else:
ctype = "application/octet-stream"
quoted = urllib.parse.quote(safe)
return ok(ctype, target.stat().st_size, {
"Content-Disposition":
"attachment; filename*=UTF-8''%s" % quoted})
return no()
def do_GET(self): # noqa: N802
RELAY.total_requests += 1
path = self.path.split("?")[0]
# 当前公网地址。故意不需要登录 ——
# 地址本来就要填到手表里，不是秘密；而且隧道换地址时
# 用户就是靠这个页面知道新地址的。
if path in ("/url", "/v1/url"):
url = public_url()
body = (url or "（还没有地址，隧道可能没起来）") + "\n"
self._send(200, body.encode("utf-8"),
"text/plain; charset=utf-8")
return
# 只读查询：这个配对码有没有被别的电脑占？
# 安装时要"先检查再决定要不要改防火墙"，而检查不能真去注册
# （一注册就等于自己占了它）。所以单独开一个只读接口。
if path in ("/v1/code/check", "/v1/check"):
query = urllib.parse.urlparse(self.path).query
wanted = (urllib.parse.parse_qs(query).get("code") or [""])[0]
wanted = wanted.strip()
if not CODE_RE.match(wanted):
self._json(400, {"ok": False, "error": "配对码要是 6 位数字"})
return
with RELAY.lock:
owner = RELAY.codes.get(wanted)
device = RELAY.devices.get(owner) if owner else None
self._json(200, {
"ok": True,
"code": wanted,
"taken": bool(owner),
"deviceName": device.name if device else "",
"online": bool(device and device.online()),
})
return
# ---- 网站图标 ----
if path in ("/favicon.ico", "/apple-touch-icon.png"):
icon = Path(__file__).resolve().parent / "favicon.ico"
if icon.is_file():
try:
blob = icon.read_bytes()
except OSError:
blob = b""
if blob:
self.send_response(200)
self.send_header("Content-Type", "image/x-icon")
self.send_header("Content-Length", str(len(blob)))
# 图标不常变，让浏览器缓存一天
self.send_header("Cache-Control", "public, max-age=86400")
self.end_headers()
self.wfile.write(blob)
return
# 没有图标文件就 404，别让请求打到后面的逻辑
self.send_response(404)
self.send_header("Content-Length", "0")
self.end_headers()
return
# ---- 成品下载（挂在管理页上给人下 APK / 整包）----
if path == "/v1/downloads":
self._json(200, {"ok": True, "files": _list_downloads()})
return
# 官网用的产品截图。
# 单独一条路由（不放进下载区），只允许 png/jpg/webp。
if path.startswith("/shot/"):
name = path[len("/shot/"):]
if (not name or ".." in name or "/" in name or "\\" in name
or not name.lower().endswith((".png", ".jpg", ".jpeg",
".webp"))):
self._json(404, {"ok": False, "error": "没有这张图"})
return
target = Path(__file__).resolve().parent / "site_shots" / name
if not target.is_file():
self._json(404, {"ok": False, "error": "没有这张图"})
return
body = target.read_bytes()
ctype = ("image/png" if name.lower().endswith(".png")
else "image/jpeg")
self.send_response(200)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "public, max-age=86400")
self.end_headers()
self.wfile.write(body)
return
if path.startswith("/download/"):
wanted = urllib.parse.unquote(path[len("/download/"):])
safe = _safe_download_name(wanted)
if not safe:
self._json(400, {"ok": False, "error": "文件名不合法"})
return
target = DOWNLOADS_DIR / safe
if not target.is_file():
self._json(404, {"ok": False, "error": "没有这个文件"})
return
try:
size = target.stat().st_size
except OSError as exc:
self._json(500, {"ok": False, "error": f"读不到: {exc}"})
return
if safe.lower().endswith(".apk"):
content_type = "application/vnd.android.package-archive"
elif safe.lower().endswith(".zip"):
content_type = "application/zip"
elif safe.lower().endswith(".txt"):
content_type = "text/plain; charset=utf-8"
else:
content_type = "application/octet-stream"
self.send_response(200)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(size))
# 文件名可能含中文，用 RFC 5987 编码
quoted = urllib.parse.quote(safe)
self.send_header(
"Content-Disposition",
f"attachment; filename*=UTF-8''{quoted}")
self.send_header("Cache-Control", "no-store")
self.end_headers()
# 分块流式发送，不要一次性 read_bytes 再 write ——
# 大文件（19MB）那样会先攒满内存、数据也迟迟不开始流动；
# 而 Cloudflare 有 100 秒的源站响应上限，越晚开始发越容易被掐。
sent = 0
try:
with target.open("rb") as stream:
while True:
chunk = stream.read(65536)
if not chunk:
break
self.wfile.write(chunk)
sent += len(chunk)
self.wfile.flush()
except (BrokenPipeError, ConnectionResetError):
# 客户端/边缘主动断开（比如大文件超时），正常现象
log(f"下载中断：{safe}（已发 {sent}/{size} 字节）")
return
except OSError as exc:
log(f"下载出错：{safe}（已发 {sent}/{size} 字节）{exc}")
return
log(f"下载：{safe}（{sent} 字节）")
return
if path in ("/v1/health", "/health"):
snap = RELAY.snapshot()
snap["service"] = "lanremote-relay"
self._json(200, snap)
return
# ---- 退出登录 ----
if path == "/admin/logout":
token = self._admin_cookie()
ADMIN_SESSIONS.pop(token, None)
self._send(302, b"", "text/html; charset=utf-8")
# 手动补 Location 和清 Cookie
return
# ---- 管理接口（GET）----
if path == "/v1/admin/netcheck":
# 网络体检：中转和 cloudflared 在同一台机器上，
# 所以能直接读它的指标，不用用户去 SSH 敲命令。
if not self._admin_gate():
self._json(401, {"ok": False, "error": "请先登录"})
return
info = {"ok": True}
# --- cloudflared 的连接数等 ---
try:
request = urllib.request.Request(
"http://127.0.0.1:20241/metrics",
headers={"User-Agent": "LanRemoteRelay/1.0"})
with urllib.request.urlopen(request, timeout=4) as response:
text = response.read(65536).decode("utf-8", "replace")
wanted = {
"cloudflared_tunnel_ha_connections": "tunnelConnections",
"cloudflared_tunnel_total_requests": "tunnelRequests",
"cloudflared_tunnel_tcp_active_sessions":
"tunnelTcpSessions",
}
for line in text.splitlines():
if line.startswith("#"):
continue
parts = line.split()
if len(parts) < 2:
continue
key = wanted.get(parts[0])
if key:
try:
info[key] = int(float(parts[1]))
except ValueError:
pass
info["tunnelMetrics"] = True
except Exception as exc: # noqa: BLE001
info["tunnelMetrics"] = False
info["tunnelError"] = f"{type(exc).__name__}"
# --- 自己这边的概况 ---
snap = RELAY.snapshot()
info["devices"] = len(snap["devices"])
info["online"] = sum(1 for d in snap["devices"] if d["online"])
info["sessions"] = snap["sessions"]
info["memoryKB"] = snap["memoryKB"]
info["uptimeSeconds"] = snap["uptimeSeconds"]
info["publicUrl"] = public_url() or ""
# --- 每条设备是不是从局域网来的 ---
lan_devices = []
for device in snap["devices"]:
for ip in (device.get("lanIps") or []):
if ip:
lan_devices.append(ip)
break
info["lanDevices"] = len(lan_devices)
self._json(200, info)
return
if path.startswith("/v1/admin/"):
if not self._admin_gate():
self._json(401, {"ok": False, "error": "请先登录"})
return
if path == "/v1/admin/devices":
self._json(200, RELAY.snapshot())
return
self._json(404, {"ok": False, "error": "没有这个管理接口"})
return
# ---- 官网首页（公开，不用登录）----
if path == "/":
self._send(200, _site_page().encode("utf-8"),
"text/html; charset=utf-8")
return
# ---- 管理后台 ----
# 挪到 /zhongzhuanzhan（用户要求）。
# /admin 保留成跳转，免得存过书签的人打不开。
if path == "/admin":
self._redirect(ADMIN_PATH)
return
if path == ADMIN_PATH:
# 管理页要登录 —— 上面会显示配对码，不能谁都能看
if not self._admin_gate():
self._send(200, _login_page(
"账号或密码不对，再试一次"
if _query(self.path, "bad") else ""
), "text/html; charset=utf-8")
return
self._admin_page()
return
self._json(404, {"ok": False, "error": "没有这个接口"})
def do_POST(self): # noqa: N802
RELAY.total_requests += 1
path = self.path.split("?")[0]
try:
self._route_post(path)
except Exception as exc: # noqa: BLE001
log(f"处理 {path} 出错：{exc}")
self._json(500, {"ok": False, "error": "服务器内部错误"})
def _route_post(self, path: str) -> None:
# ---- 被控端 ----
if path == "/v1/agent/register":
result = RELAY.register(self._json_body())
log(f"注册请求 -> ok={result.get('ok')}")
self._json(200 if result.get("ok") else 400, result)
return
if path == "/v1/agent/poll":
device = RELAY.device_by_token(self.headers.get("X-Token", ""))
if device is None:
self._json(401, {"ok": False, "error": "令牌无效"})
return
session_at_start = device.current_session
data = RELAY.agent_take(device, POLL_SECONDS)
self._send(200, data, "application/octet-stream",
extra={"X-Relay-Session": device.current_session or "-",
"X-Relay-Session-AtStart": session_at_start or "-"})
return
if path == "/v1/agent/send":
device = RELAY.device_by_token(self.headers.get("X-Token", ""))
if device is None:
self._json(401, {"ok": False, "error": "令牌无效"})
return
data = self._body()
RELAY.agent_put(device, data)
self._json(200, {"ok": True, "queued": len(data)})
return
if path == "/v1/agent/bye":
device = RELAY.device_by_token(self.headers.get("X-Token", ""))
if device is not None:
device.last_seen = 0
device.current_session = ""
log(f"被控端下线：{device.name}")
self._json(200, {"ok": True})
return
# ---- 控制端 ----
if path == "/v1/client/attach":
result = RELAY.attach(self._json_body(), self._client_ip())
self._json(200 if result.get("ok") else 403, result)
return
if path == "/v1/client/poll":
token = self.headers.get("X-Session", "")
session = RELAY.session_by_token(token)
if session is None:
if RELAY.session_replaced(token):
# 明确告诉它"你是被顶掉的"，客户端就不会抢回来了
self._json(409, {"ok": False, "replaced": True,
"error": "已被另一台手表接管"})
return
self._json(401, {"ok": False, "error": "会话无效"})
return
try:
data = RELAY.client_take(session, POLL_SECONDS)
except SessionReplaced:
self._json(409, {"ok": False, "replaced": True,
"error": "已被另一台手表接管"})
return
self._send(200, data, "application/octet-stream")
return
if path == "/v1/client/send":
token = self.headers.get("X-Session", "")
session = RELAY.session_by_token(token)
if session is None:
if RELAY.session_replaced(token):
self._json(409, {"ok": False, "replaced": True,
"error": "已被另一台手表接管"})
return
self._json(401, {"ok": False, "error": "会话无效"})
return
data = self._body()
RELAY.client_put(session, data)
self._json(200, {"ok": True, "queued": len(data)})
return
# ---- 登录 ----
if path == "/admin/login":
form = self._body().decode("utf-8", "replace")
username = ""
password = ""
from urllib.parse import unquote_plus
for piece in form.split("&"):
name, _, value = piece.partition("=")
if name == "password":
password = unquote_plus(value)
elif name == "username":
username = unquote_plus(value)
token = admin_login(username, password)
if not token:
log("管理页登录失败（%s），来自 %s"
% (username or "未填账号", self._client_ip()))
self._redirect(ADMIN_PATH + "?bad=1")
return
log("管理页登录成功（%s），来自 %s"
% (username, self._client_ip()))
self.send_response(302)
self.send_header("Location", ADMIN_PATH)
self.send_header(
"Set-Cookie",
f"lr_admin={token}; Path=/; HttpOnly; SameSite=Lax; "
f"Max-Age={ADMIN_SESSION_SECONDS}")
self.send_header("Content-Length", "0")
self.end_headers()
return
# ---- 管理接口（要登录）----
if path.startswith("/v1/admin/"):
if not self._admin_gate():
self._json(401, {"ok": False, "error": "请先登录"})
return
if path == "/v1/admin/devices":
self._json(200, RELAY.snapshot())
return
if path == "/v1/admin/kick":
device_id = str(self._json_body().get("deviceId") or "")
if not device_id:
self._json(400, {"ok": False, "error": "要给 deviceId"})
return
gone = RELAY.kick(device_id)
log(f"管理操作：踢掉 {device_id[:12]}… 的 {gone} 个会话")
self._json(200, {"ok": True, "kicked": gone})
return
if path == "/v1/admin/delete_many":
# 批量删除：勾几台删几台。
# 和"删除全部"的区别是不动没勾的，也不清孤儿绑定。
body = self._json_body()
wanted = body.get("deviceIds") or []
if not isinstance(wanted, list):
self._json(400, {"ok": False,
"error": "deviceIds 要是数组"})
return
wanted = [str(item).strip() for item in wanted if str(item).strip()]
if not wanted:
self._json(400, {"ok": False, "error": "没有勾选任何设备"})
return
results = [RELAY.delete_device(device_id=item)
for item in wanted]
done = [r for r in results if r.get("ok")]
failed = [r for r in results if not r.get("ok")]
log(f"管理操作：批量删除 {len(done)}/{len(wanted)} 台设备")
self._json(200, {
"ok": True,
"requested": len(wanted),
"deleted": len(done),
"failed": len(failed),
"devices": done,
"errors": failed[:10],
})
return
if path in ("/v1/admin/delete", "/v1/admin/delete_all"):
body = self._json_body()
if path == "/v1/admin/delete_all":
# 一次删光（清场用）
with RELAY.lock:
all_ids = list(RELAY.devices.keys())
results = [RELAY.delete_device(device_id=item)
for item in all_ids]
done = [r for r in results if r.get("ok")]
# 还要清掉**孤儿绑定** —— 设备早就下线、但配对码
# 还占着的记录。不清的话，用户想用的码可能一直
# 被一台不存在的机器占着（"我的码怎么被占了"）。
with RELAY.lock:
orphans = list(RELAY.codes.items())
RELAY.codes.clear()
RELAY._save()
log(f"管理操作：清场 —— 删除 {len(done)} 台设备，"
f"清掉 {len(orphans)} 条配对码绑定（含孤儿）")
self._json(200, {"ok": True, "deleted": len(done),
"orphansCleared": len(orphans),
"devices": done})
return
device = str(body.get("deviceId") or "").strip()
code = str(body.get("code") or "").strip()
if not device and not code:
self._json(400, {"ok": False,
"error": "要给 deviceId 或 code"})
return
result = RELAY.delete_device(device_id=device, code=code)
self._json(200 if result.get("ok") else 404, result)
return
if path == "/v1/admin/forget":
code = str(self._json_body().get("code") or "").strip()
if not CODE_RE.match(code):
self._json(400, {"ok": False, "error": "配对码要是 6 位数字"})
return
done = RELAY.forget(code)
log(f"管理操作：抹掉配对码 {code} -> {done}")
self._json(200, {"ok": True, "forgotten": done})
return
self._json(404, {"ok": False, "error": "没有这个管理接口"})
return
if path == "/v1/client/detach":
session = RELAY.session_by_token(self.headers.get("X-Session", ""))
if session is not None:
relay_sessions = getattr(RELAY, "sessions", {})
relay_sessions.pop(session.token, None)
RELAY.retired.pop(session.token, None)
self._json(200, {"ok": True})
return
self._json(404, {"ok": False, "error": "没有这个接口"})
# ---------------- 管理页 ----------------
def _admin_page(self) -> None:
"""管理后台。
实际渲染在 relay/admin.py —— 原来几百行 HTML/CSS/JS
内联在这里的 f-string 里，改一处样式要在一堆转义花括号里翻，
没法维护。拆出去之后这里只负责把数据递过去。
⚠ Content-Type 必须显式写 text/html。
_send() 默认是 application/json；不传的话浏览器会把整页
HTML 当纯文本显示出来 —— 用户看到的是一屏源码。
（这个坑真踩了：改完只测了 HTTP 200 和字节数，
没看响应头，结果页面在浏览器里是白的。）
"""
from admin import render_admin
snap = RELAY.snapshot()
self._send(200, render_admin(
snap=snap,
public=public_url() or "",
admin_path=ADMIN_PATH,
codes=len(RELAY.codes),
downloads_fn=_list_downloads,
).encode("utf-8"), "text/html; charset=utf-8")
# ======================================================================
# 下面这几个是页面/启动辅助（重做管理页时被误删，从 git 恢复）
# ======================================================================
def _esc(text) -> str:
return (str(text).replace("&", "&").replace("<", "<")
.replace(">", ">").replace('"', """))
def _kb(value: int) -> str:
if value < 1024:
return f"{value} B"
if value < 1024 * 1024:
return f"{value / 1024:.1f} KB"
return f"{value / 1024 / 1024:.1f} MB"
def _uptime(seconds: int) -> str:
days, rest = divmod(int(seconds), 86400)
hours, rest = divmod(rest, 3600)
minutes = rest // 60
if days:
return f"{days} 天 {hours} 小时"
if hours:
return f"{hours} 小时 {minutes} 分"
return f"{minutes} 分"
# ======================================================================
# 启动
# ======================================================================
class Server(ThreadingHTTPServer):
daemon_threads = True
allow_reuse_address = True
request_queue_size = 64
def main() -> int:
DATA_DIR.mkdir(parents=True, exist_ok=True)
try:
server = Server((BIND, PORT), Handler)
except OSError as exc:
print(f"监听 {BIND}:{PORT} 失败：{exc}")
return 1
log(f"中转服务器 v{VERSION} 启动，监听 {BIND}:{PORT}")
log(f"长轮询 {POLL_SECONDS:.0f} 秒 · 队列上限 "
f"{MAX_QUEUE_BYTES // 1024} KB · 数据目录 {DATA_DIR}")
key = admin_key()
log(f"管理页：http://<本机>:{PORT}/?key={key}")
log(f"（钥匙也在 {DATA_DIR / 'admin_key.txt'}）")
try:
server.serve_forever()
except KeyboardInterrupt:
log("收到中断，退出")
finally:
server.server_close()
return 0
if __name__ == "__main__":
sys.exit(main())