mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 10:10:43 +00:00
xmz7891
2061f9525f
手表端 - 功能页从 10 个杂项精简到 7 个（屏幕/鼠标/媒体/文件/电脑/语音/电源） - 删掉：游戏封锁、待办、掷骰子、抽奖、壁纸、显示器开关、定时关机、 剪贴板历史、性能/磁盘/电量、文件夹浏览、教室限制等 - 改成显式列按钮，不再把电脑端所有功能组无脑铺到表盘 - 远程鼠标控制重做：滑动区 + 左键/滚轮/右键三按钮 滚轮点按=中键、上下滑=滚动 - 新增视频/音乐播放器（手表本地播放）：进度拖动、左右滑快进退10秒、 点画面播放暂停、上一/下一、播完自动下一首 电脑端 - 新增 kill_all（结束所有进程），白名单保护系统关键进程和自身 - 新增 media_list / media_pull（播放器数据源） 安全边界：只允许媒体目录，用 relative_to 判断，防同前缀兄弟目录绕过 中转站 - known_devices 从 ID 集合改成「设备ID -> 配对码」映射 - 修复：管理员改码后关机超过 6 小时，改码会被超时清理冲掉 - 老格式 state 文件平滑升级
355 lines
13 KiB
Python
355 lines
13 KiB
Python
# -*- coding: utf-8 -*-
|
||
r'''电脑端 2.3.0 改动：结束所有进程 + 播放器数据源。
|
||
|
||
== 加的三个东西 ==
|
||
|
||
1. **`kill_all`**（一键结束所有进程）
|
||
|
||
用户要求「一键结束所有进程」。但这个操作**很容易把机器搞崩**：
|
||
杀 csrss.exe 蓝屏、杀 winlogon.exe 注销、杀 explorer.exe 桌面消失、
|
||
杀掉自己就再也连不上了。
|
||
|
||
所以这里**白名单式保护**：
|
||
* 本程序自己的 PID
|
||
* 父进程链（打包后用 pythonw 起的话，父进程被杀会连带）
|
||
* 一批系统关键进程（见 PROTECTED_NAMES）
|
||
其余全杀。
|
||
|
||
2. **`media_list`**（列出电脑上的音视频文件）
|
||
|
||
给手表的播放器用。扫这几个目录：
|
||
%USERPROFILE%\Music 音乐
|
||
%USERPROFILE%\Videos 视频
|
||
%USERPROFILE%\Downloads 下载
|
||
%USERPROFILE%\Desktop 桌面
|
||
接收目录（程序自己的）
|
||
只收常见音视频扩展名，最多 300 个。
|
||
|
||
3. **`media_pull`**（把某个媒体文件发给手表）
|
||
|
||
**安全边界**：只允许上面那几个根目录里的文件。
|
||
绝不能变成"任意路径读文件" —— 那等于给了个远程文件读取器。
|
||
解析方式：先 resolve，再检查是否在允许的根目录之内
|
||
（用 `Path.relative_to` 判断，不是简单的字符串前缀比较 ——
|
||
前缀比较会被 `C:\Users\a\Music-evil` 这种绕过）。
|
||
'''
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
FEATURES = ROOT / "windows" / "lanremote" / "features.py"
|
||
SERVER = ROOT / "windows" / "lanremote" / "server.py"
|
||
PROTOCOL = ROOT / "windows" / "lanremote" / "protocol.py"
|
||
|
||
|
||
def must(text: str, old: str, new: str, label: str) -> str:
|
||
n = text.count(old)
|
||
if n == 0:
|
||
raise SystemExit("✗ [%s] 锚点没找到" % label)
|
||
if n > 1:
|
||
raise SystemExit("✗ [%s] 锚点出现 %d 次" % (label, n))
|
||
out = text.replace(old, new, 1)
|
||
if new.strip()[:50] not in out:
|
||
raise SystemExit("✗ [%s] 替换后找不到新内容" % label)
|
||
print(" ✓ %s" % label)
|
||
return out
|
||
|
||
|
||
# ---------------------------------------------------------------- features
|
||
KILL_ALL_IMPL = '''
|
||
# ---------------- 一键结束所有进程（2.3.0）----------------
|
||
#
|
||
# 用户要求「一键结束所有进程」。这个操作很危险，所以做了**白名单保护**：
|
||
# * 本程序自己的 PID 和父进程链（杀了就再也连不上）
|
||
# * 系统关键进程（杀了会蓝屏/注销/桌面消失）
|
||
# 其余全杀。
|
||
|
||
# 绝对不能结束的进程名（小写）。
|
||
#
|
||
# 这些名字都是踩过坑或者有明确后果的：
|
||
# csrss / wininit / winlogon / services / lsass / smss -> 蓝屏或立即重启
|
||
# svchost -> 一杀就掉一片服务（网络、音频、打印全没）
|
||
# dwm -> 窗口全部失��重绘
|
||
# explorer -> 桌面和任务栏消失（虽然会自动重启，但用户会以为坏了）
|
||
# audiodg -> 音频引擎
|
||
# ykxmz / python / pythonw -> 本程序
|
||
PROTECTED_NAMES = {
|
||
"system", "system idle process", "registry", "memory compression",
|
||
"smss.exe", "csrss.exe", "wininit.exe", "winlogon.exe", "services.exe",
|
||
"lsass.exe", "lsm.exe", "svchost.exe", "dwm.exe", "fontdrvhost.exe",
|
||
"sihost.exe", "ctfmon.exe", "audiodg.exe", "spoolsv.exe",
|
||
"explorer.exe", "taskhost.exe", "taskhostw.exe", "runtimebroker.exe",
|
||
"searchindexer.exe", "wmiprvse.exe", "dllhost.exe", "conhost.exe",
|
||
"ykxmz.exe", "python.exe", "pythonw.exe", "cmd.exe", "taskkill.exe",
|
||
}
|
||
|
||
def _protected_pids(self) -> set:
|
||
"""自己和父进程链的 PID —— 这些杀了等于自断后路。"""
|
||
import os
|
||
import subprocess as _sp
|
||
pids = {os.getpid()}
|
||
pid = os.getpid()
|
||
for _ in range(6): # 往上追几层就够了
|
||
try:
|
||
out = _sp.run(
|
||
["wmic", "process", "where", "ProcessId=%d" % pid,
|
||
"get", "ParentProcessId", "/value"],
|
||
capture_output=True, timeout=10,
|
||
creationflags=getattr(_sp, "CREATE_NO_WINDOW", 0),
|
||
encoding="mbcs", errors="replace").stdout or ""
|
||
parent = 0
|
||
for line in out.splitlines():
|
||
if "=" in line and "ParentProcessId" in line:
|
||
try:
|
||
parent = int(line.split("=")[-1].strip())
|
||
except ValueError:
|
||
parent = 0
|
||
if parent <= 0 or parent in pids:
|
||
break
|
||
pids.add(parent)
|
||
pid = parent
|
||
except Exception: # noqa: BLE001
|
||
break
|
||
return pids
|
||
|
||
def _do_kill_all(self, params) -> str:
|
||
import os
|
||
me = os.getpid()
|
||
guard = set(self.PROTECTED_NAMES)
|
||
for proc in self.actions.list_processes(limit=3):
|
||
pass
|
||
protected_pids = self._protected_pids()
|
||
|
||
killed = []
|
||
skipped = 0
|
||
failed = 0
|
||
for proc in self.actions.list_processes(limit=500):
|
||
name = (proc.get("name") or "").strip()
|
||
low = name.lower()
|
||
pid = int(proc.get("pid") or 0)
|
||
if pid <= 0 or pid == me or pid in protected_pids:
|
||
skipped += 1
|
||
continue
|
||
if low in guard:
|
||
skipped += 1
|
||
continue
|
||
try:
|
||
self.actions.kill_process(pid)
|
||
killed.append(name)
|
||
except Exception: # noqa: BLE001
|
||
failed += 1
|
||
|
||
if not killed:
|
||
return "没有可结束的程序（系统和本程序都已排除）"
|
||
head = "、".join(killed[:8])
|
||
more = "" if len(killed) <= 8 else ("… 等 %d 个" % len(killed))
|
||
return ("已结束 %d 个程序：%s%s（跳过系统进程 %d 个，失败 %d 个）"
|
||
% (len(killed), head, more, skipped, failed))
|
||
'''
|
||
|
||
# ---------------------------------------------------------------- server
|
||
MEDIA_IMPL = '''
|
||
# ---------------- 播放器数据源（2.3.0）----------------
|
||
#
|
||
# 手表上的播放器要"能看进度、能左右划 seek" —— 那就必须手表自己解码播放。
|
||
# 所以电脑端负责两件事：**告诉它有哪些媒体** + **把文件发过去**。
|
||
|
||
# 允许扫描的媒体目录（相对用户主目录）。
|
||
MEDIA_SUBDIRS = ("Music", "Videos", "Downloads", "Desktop",
|
||
"音乐", "视频", "下载", "桌面")
|
||
|
||
AUDIO_EXT = {".mp3", ".wav", ".m4a", ".aac", ".flac", ".ogg",
|
||
".wma", ".amr", ".opus", ".ape"}
|
||
VIDEO_EXT = {".mp4", ".mkv", ".avi", ".mov", ".wmv", ".flv",
|
||
".webm", ".3gp", ".m4v", ".ts", ".rmvb"}
|
||
MEDIA_LIST_LIMIT = 300
|
||
|
||
def _media_roots(self) -> list:
|
||
"""允许访问的根目录（都 resolve 成绝对路径）。"""
|
||
roots = []
|
||
home = Path.home()
|
||
for sub in self.MEDIA_SUBDIRS:
|
||
p = home / sub
|
||
try:
|
||
if p.is_dir():
|
||
roots.append(p.resolve())
|
||
except OSError:
|
||
continue
|
||
try:
|
||
roots.append(Path(self.store.received_dir()).resolve())
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
return roots
|
||
|
||
def _media_list(self) -> dict:
|
||
roots = self._media_roots()
|
||
files = []
|
||
for root in roots:
|
||
try:
|
||
entries = list(root.iterdir())
|
||
except OSError:
|
||
continue
|
||
for p in entries:
|
||
try:
|
||
if not p.is_file():
|
||
continue
|
||
ext = p.suffix.lower()
|
||
if ext in self.AUDIO_EXT:
|
||
kind = "audio"
|
||
elif ext in self.VIDEO_EXT:
|
||
kind = "video"
|
||
else:
|
||
continue
|
||
files.append({
|
||
"name": p.name,
|
||
"path": str(p),
|
||
"size": p.stat().st_size,
|
||
"kind": kind,
|
||
"mtime": p.stat().st_mtime,
|
||
})
|
||
except OSError:
|
||
continue
|
||
|
||
# 新的排前面
|
||
files.sort(key=lambda f: f.get("mtime", 0), reverse=True)
|
||
total = len(files)
|
||
shown = files[:self.MEDIA_LIST_LIMIT]
|
||
return {
|
||
"ok": True,
|
||
"files": shown,
|
||
"total": total,
|
||
"shown": len(shown),
|
||
"truncated": total > len(shown),
|
||
"roots": [str(r) for r in roots],
|
||
}
|
||
|
||
def _media_pull(self, session: "Session", path: str) -> dict:
|
||
"""把一个媒体文件发给手表。
|
||
|
||
⚠ **安全边界**：只允许媒体根目录里的文件。
|
||
不能用 `str(p).startswith(root)` 这种判断 —— 会被
|
||
`C:\\Users\\me\\Music-evil\\x.mp3` 绕过（前缀相同但不是子目录）。
|
||
必须用 `relative_to`，它在路径不真的位于该目录下时会抛 ValueError。
|
||
"""
|
||
wanted = Path(str(path or "")).expanduser()
|
||
try:
|
||
wanted = wanted.resolve()
|
||
except (OSError, ValueError) as exc:
|
||
raise ActionError(f"路径不合法: {exc}") from exc
|
||
|
||
allowed = False
|
||
for root in self._media_roots():
|
||
try:
|
||
wanted.relative_to(root)
|
||
allowed = True
|
||
break
|
||
except ValueError:
|
||
continue
|
||
if not allowed:
|
||
raise ActionError("这个文件不在允许的媒体目录里")
|
||
|
||
if not wanted.is_file():
|
||
raise ActionError(f"文件不存在: {wanted.name}")
|
||
|
||
size = wanted.stat().st_size
|
||
if size <= 0:
|
||
raise ActionError("文件是空的")
|
||
if size > 512 * 1024 * 1024:
|
||
raise ActionError("文件太大（超过 512MB），播放器放不下")
|
||
|
||
blob = wanted.read_bytes()
|
||
if not session.send_chunks(P.EVT_FILE, wanted.name, blob):
|
||
raise ActionError("发送失败：连接已断开")
|
||
return {"ok": True, "name": wanted.name, "size": len(blob)}
|
||
'''
|
||
|
||
|
||
def main() -> int:
|
||
print("=" * 62)
|
||
print(" 电脑端 2.3.0 改动")
|
||
print("=" * 62)
|
||
|
||
# ---------------- features.py ----------------
|
||
print()
|
||
print("── features.py ──")
|
||
ftext = FEATURES.read_text(encoding="utf-8")
|
||
|
||
# 1) 注册 kill_all
|
||
ftext = must(
|
||
ftext,
|
||
' add("kill_games", "结束游戏进程", "action", "教室")\n',
|
||
' add("kill_games", "结束游戏进程", "action", "教室")\n'
|
||
' add("kill_all", "结束所有进程", "action", "系统")\n',
|
||
"注册 kill_all 功能")
|
||
|
||
# 2) 加实现（放在 _do_kill_games / _do_kill_by_name 之后）
|
||
anchor = (
|
||
' def _do_kill_by_name(self, params) -> str:\n'
|
||
' name = str(params.get("name") or "").strip()\n'
|
||
' if not name:\n'
|
||
' raise ValueError("进程名为空")\n'
|
||
' return self.actions.kill_process(0, name)\n'
|
||
)
|
||
ftext = must(ftext, anchor, anchor + KILL_ALL_IMPL, "kill_all 实现")
|
||
|
||
FEATURES.write_text(ftext, encoding="utf-8", newline="\n")
|
||
|
||
# ---------------- server.py ----------------
|
||
print()
|
||
print("── server.py ──")
|
||
stext = SERVER.read_text(encoding="utf-8")
|
||
|
||
# 3) 在 _file_pull 之后插入 media_* 实现
|
||
anchor = (
|
||
' def _file_pull(self, session: Session, path: str) -> dict:\n'
|
||
' target = self._resolve_received(path)\n'
|
||
' if target is None:\n'
|
||
' raise ActionError(f"文件不存在: {path}")\n'
|
||
' blob = target.read_bytes()\n'
|
||
' if not session.send_chunks(P.EVT_FILE, target.name, blob):\n'
|
||
' raise ActionError("发送文件失败：连接已断开")\n'
|
||
' return {"ok": True, "name": target.name, "size": len(blob)}\n'
|
||
)
|
||
stext = must(stext, anchor, anchor + MEDIA_IMPL, "media_list / media_pull 实现")
|
||
|
||
# 4) 路由：在 CMD_FILE_LIST / CMD_FILE_PULL 旁边加
|
||
route_anchor = (
|
||
' if cmd == P.CMD_FILE_LIST:\n'
|
||
' return self._file_list()\n'
|
||
)
|
||
stext = must(
|
||
stext,
|
||
route_anchor,
|
||
route_anchor +
|
||
' if cmd == "media_list":\n'
|
||
' return self._media_list()\n',
|
||
"路由 media_list")
|
||
|
||
pull_anchor = (
|
||
' if cmd == P.CMD_FILE_PULL:\n'
|
||
)
|
||
if stext.count(pull_anchor) != 1:
|
||
raise SystemExit("✗ [路由 media_pull] CMD_FILE_PULL 锚点不唯一")
|
||
# 找到这个 if 块的结束（下一个同级 if）
|
||
idx = stext.find(pull_anchor)
|
||
# 插在 CMD_FILE_PULL 分支之前更简单
|
||
stext = stext.replace(
|
||
pull_anchor,
|
||
' if cmd == "media_pull":\n'
|
||
' return self._media_pull(session, str(msg.get("path") or ""))\n'
|
||
+ pull_anchor,
|
||
1)
|
||
print(" ✓ 路由 media_pull")
|
||
|
||
SERVER.write_text(stext, encoding="utf-8", newline="\n")
|
||
|
||
print()
|
||
print(" 完成")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|