mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 19:30:42 +00:00
xmz7891
b62ac865ea
被控端 - 修复 MTU 问题：夹 MSS + 路径 MTU 探测（公网可用率 48% -> 100%） - 复用 HTTP 连接（免掉每次新建 TLS 的 30% 握手失败） - 每分钟回头找局域网，自动切回快 400 倍的路径 - 画面心跳帧：静止时省流量，客户端仍知道流是活的 - 锁屏时给出可操作的提示 手表端 - 内置 GTS Root R4 / GlobalSign Root CA，修 SSLHandshakeException - 连接错误翻译成人话（证书/时间/解析/超时分别提示） - 经中转连上后用服务器报的地址刷新已配对记录 - 新增局域网连接页面、服务器延迟检测 服务器端 - 支持 chunked 请求体（修每 16 秒一次的连接错乱与大面积 502） - 网络调优脚本 net-tune.sh（开机自动应用） - 管理页：批量删除、网络体检、下载区 - 隧道改 HTTP/2 + 双连接 文档 - 许可协议 / 安装前必读 / 安装后须知（作者：小麦子）
136 lines
5.2 KiB
Python
136 lines
5.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""把功能注册表接到服务端。
|
|
|
|
* 新增 feature_list / feature_run 两条命令
|
|
* RemoteServer 里挂一个 Features 实例
|
|
* 顺手补一个洞：_resolve_received 原来接受**任意绝对路径**，
|
|
等于能拉取电脑上任何文件。现在限定在接收目录 + 允许的用户目录内。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote\windows\lanremote")
|
|
SERVER = ROOT / "server.py"
|
|
PROTOCOL = ROOT / "protocol.py"
|
|
|
|
PROTO_OLD = 'CMD_SAY_CLIP = "say_clip" # 朗读剪贴板\n'
|
|
PROTO_NEW = ('CMD_SAY_CLIP = "say_clip" # 朗读剪贴板\n'
|
|
'CMD_FEATURE_LIST = "feature_list" # 电脑端能做哪些功能\n'
|
|
'CMD_FEATURE_RUN = "feature_run" # 执行某个功能\n')
|
|
|
|
IMPORT_OLD = "from .discovery import DISCOVERY_PORT, DiscoveryResponder\n"
|
|
IMPORT_NEW = ("from .discovery import DISCOVERY_PORT, DiscoveryResponder\n"
|
|
"from .features import Features\n")
|
|
|
|
INIT_OLD = " self._total_connections = 0\n"
|
|
INIT_NEW = (" self._total_connections = 0\n"
|
|
" # 功能注册表（夜间模式 / 亮度 / 文件浏览…）\n"
|
|
" self.features: Features | None = None\n")
|
|
|
|
# 在 actions 赋值处挂上 features —— 找 server 里 actions 的设置点
|
|
ACTIONS_ANCHOR = " self.actions = actions\n"
|
|
ACTIONS_NEW = (" self.actions = actions\n"
|
|
" self.features = Features(actions) if actions else None\n")
|
|
|
|
CMD_ANCHOR = ' if cmd == P.CMD_SET_VOICE:'
|
|
CMD_NEW = ''' if cmd == P.CMD_FEATURE_LIST:
|
|
if self.features is None:
|
|
raise ActionError("功能不可用")
|
|
return {"ok": True, "features": self.features.defs(),
|
|
"extra_apps": self.features.extra_apps(),
|
|
"extra_commands": self.features.extra_commands(),
|
|
"roots": [str(p) for p in self.features.allowed_roots()]}
|
|
|
|
if cmd == P.CMD_FEATURE_RUN:
|
|
if self.features is None:
|
|
raise ActionError("功能不可用")
|
|
name = str(msg.get("name") or "")
|
|
params = msg.get("params") or {}
|
|
if not isinstance(params, dict):
|
|
params = {}
|
|
try:
|
|
note = self.features.run(name, params)
|
|
except ValueError as exc:
|
|
raise ActionError(str(exc)) from exc
|
|
self._log(f"[功能] {name} -> {note}")
|
|
return {"ok": True, "note": str(note)[:4000]}
|
|
|
|
if cmd == P.CMD_SET_VOICE:'''
|
|
|
|
# 收紧 _resolve_received 的绝对路径
|
|
RESOLVE_OLD = ''' raw = Path(path)
|
|
if raw.is_absolute() and raw.is_file():
|
|
return raw
|
|
'''
|
|
RESOLVE_NEW = ''' raw = Path(path)
|
|
if raw.is_absolute() and raw.is_file():
|
|
# 只允许拉取「接收目录」或「用户自己的目录」里的文件。
|
|
# 否则这条命令等于"能下载电脑上任意文件"，是个大洞。
|
|
try:
|
|
resolved = raw.resolve()
|
|
except OSError:
|
|
return None
|
|
allowed = [self.store.received_dir()]
|
|
if self.features is not None:
|
|
allowed.extend(self.features.allowed_roots())
|
|
for root in allowed:
|
|
try:
|
|
resolved.relative_to(root.resolve())
|
|
return resolved
|
|
except (ValueError, OSError):
|
|
continue
|
|
self._log(f"[文件] 拒绝拉取允许范围外的路径: {raw}")
|
|
return None
|
|
'''
|
|
|
|
|
|
def main() -> int:
|
|
ok = True
|
|
|
|
text = PROTOCOL.read_text(encoding="utf-8")
|
|
if PROTO_OLD not in text:
|
|
print(" [失败] protocol.py 锚点未匹配")
|
|
ok = False
|
|
else:
|
|
text = text.replace(PROTO_OLD, PROTO_NEW, 1)
|
|
PROTOCOL.write_text(text, encoding="utf-8")
|
|
print(" protocol.py: 新增 2 条命令")
|
|
|
|
stext = SERVER.read_text(encoding="utf-8")
|
|
for old, new, tag in ((IMPORT_OLD, IMPORT_NEW, "导入"),
|
|
(INIT_OLD, INIT_NEW, "初始化"),
|
|
(CMD_ANCHOR, CMD_NEW, "命令处理"),
|
|
(RESOLVE_OLD, RESOLVE_NEW, "收紧文件路径")):
|
|
if old not in stext:
|
|
print(f" [失败] server.py {tag} 锚点未匹配")
|
|
ok = False
|
|
else:
|
|
stext = stext.replace(old, new, 1)
|
|
print(f" server.py / {tag}: OK")
|
|
|
|
if ACTIONS_ANCHOR in stext:
|
|
stext = stext.replace(ACTIONS_ANCHOR, ACTIONS_NEW, 1)
|
|
print(" server.py / 挂载 Features: OK")
|
|
else:
|
|
print(" [警告] 没找到 actions 赋值点，features 需要在 run_agent 里挂")
|
|
|
|
if ok:
|
|
SERVER.write_text(stext, encoding="utf-8")
|
|
|
|
print("\n语法检查：")
|
|
import ast
|
|
for name, path in (("protocol.py", PROTOCOL), ("server.py", SERVER)):
|
|
try:
|
|
ast.parse(path.read_text(encoding="utf-8"))
|
|
print(f" {name}: OK")
|
|
except SyntaxError as exc:
|
|
print(f" {name}: 语法错误 {exc}")
|
|
ok = False
|
|
return 0 if ok else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|