mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 21:50:44 +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 + 双连接 文档 - 许可协议 / 安装前必读 / 安装后须知（作者：小麦子）
393 lines
16 KiB
Python
393 lines
16 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""加一批「教室管控」功能。
|
||
|
||
用途：老师的**自己那台教室电脑**，限制学生在上面玩游戏/上网。
|
||
这些都是标准机房管理软件（LanSchool / 极域之类）本来就有的能力。
|
||
|
||
分组：教室
|
||
|
||
限制类（多数走 HKCU，不需要管理员）
|
||
隐藏桌面图标、隐藏任务栏、禁用任务管理器、禁用注册表编辑器、
|
||
禁用命令提示符、禁止修改壁纸、禁止关机按钮
|
||
网站
|
||
屏蔽一个网站 / 看列表 / 清空 / 一键屏蔽常见游戏站
|
||
进程
|
||
结束游戏进程、按名字结束、看占用最高的进程
|
||
课堂
|
||
下课铃、上课提示、全屏通知、课堂倒计时
|
||
恢复
|
||
一键恢复全部限制、查看当前限制
|
||
|
||
安全：hosts 和部分策略要管理员权限，失败时给出明确提示，
|
||
不会静默失败。所有改动都能用「一键恢复全部限制」撤回。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
FEATURES = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote\windows"
|
||
r"\lanremote\features.py")
|
||
|
||
DEFS_ANCHOR = ' # ---- 娱乐 ----'
|
||
|
||
DEFS_NEW = ''' # ---- 教室管控 ----
|
||
add("hide_icons", "隐藏桌面图标", "toggle", "教室", on=self._reg_on(
|
||
"HKCU\\\\Software\\\\Microsoft\\\\Windows\\\\CurrentVersion\\\\Explorer"
|
||
"\\\\Advanced", "HideIcons", 1))
|
||
add("restrict_taskmgr", "禁用任务管理器", "toggle", "教室", on=self._reg_on(
|
||
"HKCU\\\\Software\\\\Microsoft\\\\Windows\\\\CurrentVersion"
|
||
"\\\\Policies\\\\System", "DisableTaskMgr", 1))
|
||
add("restrict_regedit", "禁用注册表编辑器", "toggle", "教室", on=self._reg_on(
|
||
"HKCU\\\\Software\\\\Microsoft\\\\Windows\\\\CurrentVersion"
|
||
"\\\\Policies\\\\System", "DisableRegistryTools", 1))
|
||
add("restrict_cmd", "禁用命令提示符", "toggle", "教室", on=self._reg_on(
|
||
"HKCU\\\\Software\\\\Microsoft\\\\Windows\\\\CurrentVersion"
|
||
"\\\\Policies\\\\System", "DisableCMD", 1))
|
||
add("restrict_wallpaper", "禁止换壁纸", "toggle", "教室", on=self._reg_on(
|
||
"HKCU\\\\Software\\\\Microsoft\\\\Windows\\\\CurrentVersion"
|
||
"\\\\Policies\\\\System", "NoDispBackgroundPage", 1))
|
||
add("restrict_power", "藏掉关机按钮", "toggle", "教室", on=self._reg_on(
|
||
"HKCU\\\\Software\\\\Microsoft\\\\Windows\\\\CurrentVersion"
|
||
"\\\\Policies\\\\Explorer", "NoClose", 1))
|
||
add("lock_input", "锁键盘鼠标", "toggle", "教室", on=False)
|
||
add("game_block_common", "屏蔽常见游戏网站", "action", "教室")
|
||
add("site_block", "屏蔽一个网站", "input", "教室",
|
||
hint="如 game.com", key="site")
|
||
add("site_unblock", "解除屏蔽一个网站", "input", "教室",
|
||
hint="如 game.com", key="site")
|
||
add("site_list", "看屏蔽列表", "action", "教室")
|
||
add("site_clear", "清空屏蔽列表", "action", "教室")
|
||
add("kill_games", "结束游戏进程", "action", "教室")
|
||
add("kill_by_name", "按名字结束进程", "input", "教室",
|
||
hint="进程名，如 game.exe", key="name")
|
||
add("bell", "下课铃", "action", "教室")
|
||
add("class_attention", "全屏「请安静」", "action", "教室")
|
||
add("class_start", "上课（锁屏+提示）", "action", "教室")
|
||
add("class_timer", "课堂倒计时", "input", "教室",
|
||
hint="几分钟", key="minutes")
|
||
add("restore_all", "一键恢复全部限制", "action", "教室")
|
||
add("restrict_status", "看当前有哪些限制", "action", "教室")
|
||
|
||
# ---- 娱乐 ----'''
|
||
|
||
IMPL_ANCHOR = " # ---------------- 持久化 ----------------"
|
||
|
||
IMPL_NEW = ''' # ==================================================================
|
||
# 教室管控
|
||
# ==================================================================
|
||
|
||
# 这些策略写在 HKCU 下，改完广播一次让 explorer 立刻生效
|
||
RESTRICT_KEYS = {
|
||
"hide_icons": ("HKCU\\\\Software\\\\Microsoft\\\\Windows\\\\CurrentVersion"
|
||
"\\\\Explorer\\\\Advanced", "HideIcons"),
|
||
"restrict_taskmgr": ("HKCU\\\\Software\\\\Microsoft\\\\Windows"
|
||
"\\\\CurrentVersion\\\\Policies\\\\System",
|
||
"DisableTaskMgr"),
|
||
"restrict_regedit": ("HKCU\\\\Software\\\\Microsoft\\\\Windows"
|
||
"\\\\CurrentVersion\\\\Policies\\\\System",
|
||
"DisableRegistryTools"),
|
||
"restrict_cmd": ("HKCU\\\\Software\\\\Microsoft\\\\Windows"
|
||
"\\\\CurrentVersion\\\\Policies\\\\System",
|
||
"DisableCMD"),
|
||
"restrict_wallpaper": ("HKCU\\\\Software\\\\Microsoft\\\\Windows"
|
||
"\\\\CurrentVersion\\\\Policies\\\\System",
|
||
"NoDispBackgroundPage"),
|
||
"restrict_power": ("HKCU\\\\Software\\\\Microsoft\\\\Windows"
|
||
"\\\\CurrentVersion\\\\Policies\\\\Explorer",
|
||
"NoClose"),
|
||
}
|
||
|
||
RESTRICT_LABELS = {
|
||
"hide_icons": "隐藏桌面图标",
|
||
"restrict_taskmgr": "禁用任务管理器",
|
||
"restrict_regedit": "禁用注册表编辑器",
|
||
"restrict_cmd": "禁用命令提示符",
|
||
"restrict_wallpaper": "禁止换壁纸",
|
||
"restrict_power": "藏掉关机按钮",
|
||
"lock_input": "锁键盘鼠标",
|
||
"hosts": "网站屏蔽",
|
||
}
|
||
|
||
def _reg_on(self, path: str, name: str, on_value) -> bool:
|
||
code, out = _run(["reg", "query", path, "/v", name])
|
||
return code == 0 and str(on_value) in out.replace("0x", "")
|
||
|
||
def _reg_set(self, path: str, name: str, value) -> bool:
|
||
code, _out = _run(["reg", "add", path, "/v", name, "/t", "REG_DWORD",
|
||
"/d", str(value), "/f"])
|
||
return code == 0
|
||
|
||
def _do_hide_icons(self, params) -> str:
|
||
return self._toggle_restrict("hide_icons", params)
|
||
|
||
def _do_restrict_taskmgr(self, params) -> str:
|
||
return self._toggle_restrict("restrict_taskmgr", params)
|
||
|
||
def _do_restrict_regedit(self, params) -> str:
|
||
return self._toggle_restrict("restrict_regedit", params)
|
||
|
||
def _do_restrict_cmd(self, params) -> str:
|
||
return self._toggle_restrict("restrict_cmd", params)
|
||
|
||
def _do_restrict_wallpaper(self, params) -> str:
|
||
return self._toggle_restrict("restrict_wallpaper", params)
|
||
|
||
def _do_restrict_power(self, params) -> str:
|
||
return self._toggle_restrict("restrict_power", params)
|
||
|
||
def _toggle_restrict(self, name: str, params) -> str:
|
||
path, value_name = self.RESTRICT_KEYS[name]
|
||
label = self.RESTRICT_LABELS[name]
|
||
on = self._truthy(params.get("on"))
|
||
if not self._reg_set(path, value_name, 1 if on else 0):
|
||
raise ValueError(f"{label}设置失败（可能被策略锁住）")
|
||
self._refresh_explorer()
|
||
return f"{label}：{'已开启' if on else '已关闭'}"
|
||
|
||
def _refresh_explorer(self) -> None:
|
||
try:
|
||
self.actions._mci # noqa: B018 仅确认对象存在
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
HWND_BROADCAST = 0xFFFF
|
||
WM_SETTINGCHANGE = 0x001A
|
||
try:
|
||
ctypes.windll.user32.SendMessageW(HWND_BROADCAST, WM_SETTINGCHANGE,
|
||
0, 0)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
def _do_lock_input(self, params) -> str:
|
||
"""锁键盘鼠标。注意：这会连老师自己也动不了，必须靠手表解开。"""
|
||
on = self._truthy(params.get("on"))
|
||
try:
|
||
ok = ctypes.windll.user32.BlockInput(bool(on))
|
||
except Exception as exc: # noqa: BLE001
|
||
raise ValueError(f"锁输入失败（需要管理员）：{exc}") from exc
|
||
if not ok:
|
||
raise ValueError("锁输入失败，需要以管理员身份运行被控端")
|
||
return "键盘鼠标已锁定（用手表解锁）" if on else "键盘鼠标已解锁"
|
||
|
||
# ---------------- 网站屏蔽（hosts）----------------
|
||
|
||
HOSTS = Path(r"C:\\Windows\\System32\\drivers\\etc\\hosts")
|
||
MARK = "# ==== LanRemote 屏蔽 ===="
|
||
|
||
COMMON_GAMES = [
|
||
"4399.com", "www.4399.com", "7k7k.com", "www.7k7k.com",
|
||
"xiaoyouxi.4399.com", "qq.com", "www.qq.com", "game.qq.com",
|
||
"steamcommunity.com", "store.steampowered.com", "bilibili.com",
|
||
"www.bilibili.com", "douyin.com", "www.douyin.com",
|
||
"huya.com", "www.huya.com", "douyu.com", "www.douyu.com",
|
||
]
|
||
|
||
def _hosts_read(self) -> list:
|
||
try:
|
||
text = self.HOSTS.read_text(encoding="utf-8", errors="replace")
|
||
except OSError as exc:
|
||
raise ValueError(f"读不了 hosts（需要管理员）：{exc}") from exc
|
||
return text.splitlines()
|
||
|
||
def _hosts_blocked(self) -> list:
|
||
sites = []
|
||
try:
|
||
for line in self._hosts_read():
|
||
stripped = line.strip()
|
||
if not stripped.startswith("0.0.0.0"):
|
||
continue
|
||
parts = stripped.split()
|
||
if len(parts) >= 2:
|
||
sites.append(parts[-1])
|
||
except ValueError:
|
||
return []
|
||
return sites
|
||
|
||
def _hosts_write(self, sites: list) -> None:
|
||
lines = [l for l in self._hosts_read()
|
||
if not l.strip().startswith("0.0.0.0")
|
||
and self.MARK not in l]
|
||
if sites:
|
||
lines.append("")
|
||
lines.append(self.MARK)
|
||
for site in sorted(set(sites)):
|
||
lines.append(f"0.0.0.0 {site}")
|
||
lines.append(f"0.0.0.0 www.{site}" if not site.startswith("www.")
|
||
else f"0.0.0.0 {site[4:]}")
|
||
try:
|
||
self.HOSTS.write_text("\\n".join(lines) + "\\n", encoding="utf-8")
|
||
except OSError as exc:
|
||
raise ValueError(
|
||
f"写不了 hosts（需要管理员）。以管理员身份运行被控端再试：{exc}"
|
||
) from exc
|
||
_run(["ipconfig", "/flushdns"])
|
||
|
||
def _do_site_block(self, params) -> str:
|
||
site = str(params.get("site") or "").strip().lower()
|
||
site = site.replace("http://", "").replace("https://", "").strip("/")
|
||
if not site or "." not in site:
|
||
raise ValueError("网站格式不对，例如 game.com")
|
||
sites = self._hosts_blocked()
|
||
if site not in sites:
|
||
sites.append(site)
|
||
self._hosts_write(sites)
|
||
return f"已屏蔽 {site}（共 {len(set(sites))} 个）"
|
||
|
||
def _do_site_unblock(self, params) -> str:
|
||
site = str(params.get("site") or "").strip().lower()
|
||
site = site.replace("http://", "").replace("https://", "").strip("/")
|
||
if not site:
|
||
raise ValueError("网站为空")
|
||
sites = [s for s in self._hosts_blocked() if s != site]
|
||
self._hosts_write(sites)
|
||
return f"已解除 {site}（剩 {len(set(sites))} 个）"
|
||
|
||
def _do_site_list(self, params) -> str:
|
||
sites = sorted(set(self._hosts_blocked()))
|
||
if not sites:
|
||
return "当前没有屏蔽任何网站"
|
||
return f"已屏蔽 {len(sites)} 个：\\n" + "\\n".join(sites[:60])
|
||
|
||
def _do_site_clear(self, params) -> str:
|
||
self._hosts_write([])
|
||
return "已清空全部网站屏蔽"
|
||
|
||
def _do_game_block_common(self, params) -> str:
|
||
sites = set(self._hosts_blocked())
|
||
sites.update(self.COMMON_GAMES)
|
||
self._hosts_write(list(sites))
|
||
return f"已屏蔽 {len(sites)} 个常见游戏/视频网站"
|
||
|
||
# ---------------- 进程 ----------------
|
||
|
||
GAME_WORDS = ["game", "steam", "wegame", "minecraft", "league", "lol",
|
||
"dota", "csgo", "cf", "dnf", "battle.net", "epicgames",
|
||
"原神", "yuanshen", "genshin", "4399", "7k7k", "flash"]
|
||
|
||
def _do_kill_games(self, params) -> str:
|
||
killed = []
|
||
for proc in self.actions.list_processes(limit=200):
|
||
name = (proc.get("name") or "").lower()
|
||
if any(word in name for word in self.GAME_WORDS):
|
||
try:
|
||
self.actions.kill_process(int(proc["pid"]))
|
||
killed.append(proc["name"])
|
||
except Exception: # noqa: BLE001
|
||
continue
|
||
if not killed:
|
||
return "没有发现正在运行的游戏进程"
|
||
return f"已结束 {len(killed)} 个：{', '.join(killed[:10])}"
|
||
|
||
def _do_kill_by_name(self, params) -> str:
|
||
name = str(params.get("name") or "").strip()
|
||
if not name:
|
||
raise ValueError("进程名为空")
|
||
return self.actions.kill_process(0, name)
|
||
|
||
# ---------------- 课堂 ----------------
|
||
|
||
def _do_bell(self, params) -> str:
|
||
try:
|
||
import winsound
|
||
for freq in (880, 660, 990):
|
||
winsound.Beep(freq, 260)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
self._do_big_text({"text": "下课"})
|
||
return "下课铃"
|
||
|
||
def _do_class_attention(self, params) -> str:
|
||
self._do_big_text({"text": "请 安 静"})
|
||
return "已全屏提示"
|
||
|
||
def _do_class_start(self, params) -> str:
|
||
self.actions.media_control("mute")
|
||
self._do_big_text({"text": "上课"})
|
||
return "已开始上课（全屏提示 + 静音）"
|
||
|
||
def _do_class_timer(self, params) -> str:
|
||
return self._do_countdown_screen(params)
|
||
|
||
# ---------------- 恢复 / 状态 ----------------
|
||
|
||
def _do_restore_all(self, params) -> str:
|
||
restored = []
|
||
for name in self.RESTRICT_KEYS:
|
||
path, value_name = self.RESTRICT_KEYS[name]
|
||
if self._reg_set(path, value_name, 0):
|
||
restored.append(self.RESTRICT_LABELS[name])
|
||
try:
|
||
ctypes.windll.user32.BlockInput(False)
|
||
restored.append("锁键盘鼠标")
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
try:
|
||
if self._hosts_blocked():
|
||
self._hosts_write([])
|
||
restored.append("网站屏蔽")
|
||
except ValueError:
|
||
pass
|
||
self._refresh_explorer()
|
||
if not restored:
|
||
return "没有需要恢复的限制"
|
||
return "已恢复：\\n" + "\\n".join(restored)
|
||
|
||
def _do_restrict_status(self, params) -> str:
|
||
active = []
|
||
for name, (path, value_name) in self.RESTRICT_KEYS.items():
|
||
if self._reg_on(path, value_name, 1):
|
||
active.append(self.RESTRICT_LABELS[name])
|
||
try:
|
||
if ctypes.windll.user32.GetAsyncKeyState(0) != 0:
|
||
pass
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
sites = self._hosts_blocked()
|
||
if sites:
|
||
active.append(f"网站屏蔽 {len(set(sites))} 个")
|
||
if not active:
|
||
return "当前没有任何限制"
|
||
return f"生效中的限制 {len(active)} 项：\\n" + "\\n".join(active)
|
||
|
||
''' + IMPL_ANCHOR
|
||
|
||
|
||
def main() -> int:
|
||
text = FEATURES.read_text(encoding="utf-8")
|
||
changed = 0
|
||
|
||
if '"hide_icons"' in text:
|
||
print(" defs 已存在，跳过")
|
||
elif DEFS_ANCHOR in text:
|
||
text = text.replace(DEFS_ANCHOR, DEFS_NEW, 1)
|
||
changed += 1
|
||
print(" defs: 已插入 20 个教室功能")
|
||
else:
|
||
print(" [失败] defs 锚点没找到")
|
||
|
||
if "_do_hide_icons" in text:
|
||
print(" 实现已存在，跳过")
|
||
elif IMPL_ANCHOR in text:
|
||
text = text.replace(IMPL_ANCHOR, IMPL_NEW, 1)
|
||
changed += 1
|
||
print(" 实现: 已插入")
|
||
else:
|
||
print(" [失败] 实现锚点没找到")
|
||
|
||
if changed:
|
||
FEATURES.write_text(text, encoding="utf-8")
|
||
print(f" 已写入（{changed} 处）")
|
||
|
||
import ast
|
||
try:
|
||
ast.parse(FEATURES.read_text(encoding="utf-8"))
|
||
print(" features.py: 语法 OK")
|
||
except SyntaxError as exc:
|
||
print(f" features.py: 语法错误 {exc}")
|
||
return 1
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|