JOJO
1a4e5baa23
威胁模型：拥有普通账号的已登录用户攻击服务器。完整审计报告见
_experiments/security_audit_2026-09-02/（不入库），含实测复现记录。
严重/高危修复：
- 彻底移除文件夹打包下载（/api/download/folder 端点删除，gui/api_v1
目录下载分支改 410）：文件型符号链接会被宿主进程解析，实测可读宿主
任意文件（含 settings.json 全部 LLM 密钥），删除比补校验更彻底
- XFF 伪造防护：get_client_ip 仅信任 ASTRION_TRUSTED_PROXIES
（默认空=不信 XFF）；登录新增账号维度锁定（5 次失败锁 300s）；
注册邀请码加爆破锁定；限流桶/失败表加 2 万键上限+GC 回收
- auth_debug.log 接入大小轮转；/api/client_debug_log 加限流+长度截断
- /host-login 增加 LINUX_SAFETY 检查且仅回环地址可用
容器加固：
- 默认 cpus=1 / memory=1g，新增 pids-limit=512、memory-swap、
no-new-privileges
- 新增每用户容器配额 MAX_ACTIVE_CONTAINERS_PER_USER（默认 3），
防止单用户占满全局容器池
api_v1：
- prompts/personalizations 的 name 加白名单校验（^[A-Za-z0-9_-]{1,64}$），
修复路径穿越写入；对话元数据中的引用名同步加白名单
- workspaces/conversations/messages/upload 四端点加用户维度限流
- 消息体加 MAX_MESSAGE_CHARS 上限（默认 200000，/api/tasks 同步）
- _path_within 加分隔符边界（修复 startswith 前缀碰撞）
其他：
- SVG 预览强制 application/octet-stream（修存储型 XSS 漏网）
- monitor_snapshot 缓存键加 username 维度（修跨用户快照读取）
- ensure/delete_workspace 加 workspace_id 白名单+父目录二次核验
- delete_folder 拒绝删除工作区根（crud_mixin 与容器代理同步修）
- GuiFileManager 死代码加名称校验防复活
- admin_dashboard 静态壳非 admin 访问一律 404
- /api/app/apk/latest 加登录校验+限流（原未认证可拉 133MB）
有意未修（见报告 7.3 遗留清单）：容器 egress 过滤（部署层）、
--cap-drop ALL（怕破坏容器内工作流，已先上 no-new-privileges）、
API 用户/子智能体 LLM 按 token 计费（待产品决策）、str(exc) 收口、
WebSocket 限流、network_permission=restricted 容器语义。
验证：全部文件 py_compile 通过；冒烟测试 6/6 通过；关键新逻辑
（名称校验/路径边界/容器代理删除防护/限流回收）已单测级验证；
端点级行为待服务重启后实测复核（清单见报告 7.5）。
Co-authored-by: Astrion powered by Kimi-K3 <astrion-agent@users.noreply.github.com>
346 lines
12 KiB
Python
346 lines
12 KiB
Python
# modules/gui_file_manager.py - GUI 文件管理专用服务
|
|
|
|
import os
|
|
import shutil
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional, Tuple
|
|
|
|
from modules.i18n import tr
|
|
|
|
|
|
@dataclass
|
|
class FileEntry:
|
|
"""用于前端显示的文件/目录条目"""
|
|
|
|
name: str
|
|
path: str
|
|
type: str # file / directory
|
|
size: int
|
|
modified_at: float
|
|
extension: Optional[str]
|
|
is_editable: bool
|
|
|
|
|
|
class GuiFileManager:
|
|
"""面向 GUI 的文件管理器，实现桌面式操作所需的能力"""
|
|
|
|
EDITABLE_EXTENSIONS = {
|
|
".txt",
|
|
".md",
|
|
".py",
|
|
".js",
|
|
".ts",
|
|
".json",
|
|
".yaml",
|
|
".yml",
|
|
".html",
|
|
".css",
|
|
".scss",
|
|
".less",
|
|
".xml",
|
|
".csv",
|
|
".ini",
|
|
".cfg",
|
|
".toml",
|
|
".sh",
|
|
".bat",
|
|
".java",
|
|
".kt",
|
|
".go",
|
|
".rs",
|
|
".c",
|
|
".cpp",
|
|
".h",
|
|
".hpp",
|
|
".vue",
|
|
".svelte",
|
|
".php",
|
|
".rb",
|
|
".swift",
|
|
".dart",
|
|
".sql",
|
|
}
|
|
|
|
MAX_TEXT_FILE_SIZE = 2 * 1024 * 1024 # 2MB
|
|
|
|
def __init__(self, base_path: str):
|
|
self.base_path = Path(base_path).expanduser().resolve()
|
|
if not self.base_path.exists():
|
|
raise ValueError("Base path does not exist")
|
|
|
|
# -------------------------
|
|
# 路径解析与安全检查
|
|
# -------------------------
|
|
|
|
def _resolve(self, relative: Optional[str]) -> Path:
|
|
relative = (relative or "").strip()
|
|
target = (self.base_path if not relative else (self.base_path / relative)).resolve()
|
|
try:
|
|
target.relative_to(self.base_path)
|
|
except ValueError:
|
|
raise ValueError(tr("gui_file.path_escape"))
|
|
return target
|
|
|
|
def _to_relative(self, absolute: Path) -> str:
|
|
return str(absolute.relative_to(self.base_path)).replace("\\", "/")
|
|
|
|
def _unique_name(self, directory: Path, name: str) -> Path:
|
|
candidate = directory / name
|
|
if not candidate.exists():
|
|
return candidate
|
|
|
|
stem = candidate.stem
|
|
suffix = candidate.suffix
|
|
counter = 1
|
|
while candidate.exists():
|
|
candidate = directory / f"{stem}_copy{counter if counter > 1 else ''}{suffix}"
|
|
counter += 1
|
|
return candidate
|
|
|
|
def _check_editable(self, path: Path) -> bool:
|
|
if not path.is_file():
|
|
return False
|
|
if path.stat().st_size > self.MAX_TEXT_FILE_SIZE:
|
|
return False
|
|
ext = path.suffix.lower()
|
|
if ext in self.EDITABLE_EXTENSIONS:
|
|
return True
|
|
# 无扩展名的文本文件尝试 UTF-8 读取判断
|
|
if not ext:
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as fh:
|
|
fh.read(2048)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
return False
|
|
|
|
# -------------------------
|
|
# 列表与元数据
|
|
# -------------------------
|
|
|
|
def list_directory(self, relative: Optional[str] = None) -> Tuple[str, List[FileEntry]]:
|
|
directory = self._resolve(relative)
|
|
if not directory.exists():
|
|
raise FileNotFoundError(tr("gui_file.dir_not_found"))
|
|
if not directory.is_dir():
|
|
raise NotADirectoryError(tr("gui_file.not_a_directory"))
|
|
|
|
entries: List[FileEntry] = []
|
|
for entry in sorted(directory.iterdir(), key=lambda p: (p.is_file(), p.name.lower())):
|
|
stat = entry.stat()
|
|
entries.append(
|
|
FileEntry(
|
|
name=entry.name,
|
|
path=self._to_relative(entry),
|
|
type="directory" if entry.is_dir() else "file",
|
|
size=stat.st_size,
|
|
modified_at=stat.st_mtime,
|
|
extension=entry.suffix.lower() if entry.is_file() else None,
|
|
is_editable=self._check_editable(entry),
|
|
)
|
|
)
|
|
relative_path = "" if directory == self.base_path else self._to_relative(directory)
|
|
return relative_path, entries
|
|
|
|
def breadcrumb(self, relative: Optional[str]) -> List[Dict[str, str]]:
|
|
crumbs: List[Dict[str, str]] = []
|
|
directory = self._resolve(relative)
|
|
try:
|
|
directory.relative_to(self.base_path)
|
|
except ValueError:
|
|
raise ValueError(tr("gui_file.path_escape"))
|
|
|
|
current = directory
|
|
while True:
|
|
relative_path = "" if current == self.base_path else self._to_relative(current)
|
|
crumbs.append({"name": current.name if current != self.base_path else "根目录", "path": relative_path})
|
|
if current == self.base_path:
|
|
break
|
|
current = current.parent
|
|
crumbs.reverse()
|
|
return crumbs
|
|
|
|
# -------------------------
|
|
# 基本操作
|
|
# -------------------------
|
|
|
|
def _sanitize_entry_name(self, name: str) -> str:
|
|
"""条目名称段级校验：禁止路径分隔符与穿越片段（防 `../../` 越出工作区）。"""
|
|
sanitized = (name or "").strip()
|
|
if not sanitized:
|
|
raise ValueError(tr("gui_file.name_empty"))
|
|
if sanitized in {".", ".."} or "/" in sanitized or "\\" in sanitized or "\x00" in sanitized:
|
|
raise ValueError(tr("gui_file.name_empty"))
|
|
return sanitized
|
|
|
|
def create_entry(self, parent_relative: Optional[str], name: str, entry_type: str) -> str:
|
|
parent = self._resolve(parent_relative)
|
|
if not parent.exists():
|
|
raise FileNotFoundError(tr("gui_file.parent_not_found"))
|
|
if not parent.is_dir():
|
|
raise NotADirectoryError(tr("gui_file.parent_not_directory"))
|
|
|
|
sanitized = self._sanitize_entry_name(name)
|
|
|
|
target = parent / sanitized
|
|
if target.exists():
|
|
raise FileExistsError(tr("gui_file.name_exists"))
|
|
|
|
if entry_type == "directory":
|
|
target.mkdir(parents=False, exist_ok=False)
|
|
elif entry_type == "file":
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.touch()
|
|
else:
|
|
raise ValueError(tr("gui_file.unsupported_type"))
|
|
|
|
return self._to_relative(target)
|
|
|
|
def delete_entries(self, relative_paths: List[str]) -> Dict[str, str]:
|
|
results: Dict[str, str] = {}
|
|
for rel in relative_paths:
|
|
target = self._resolve(rel)
|
|
if not target.exists():
|
|
results[rel] = "missing"
|
|
continue
|
|
try:
|
|
if target.is_dir():
|
|
shutil.rmtree(target)
|
|
else:
|
|
target.unlink()
|
|
results[rel] = "deleted"
|
|
except Exception as exc:
|
|
results[rel] = f"error: {exc}"
|
|
return results
|
|
|
|
def rename_entry(self, relative_path: str, new_name: str) -> str:
|
|
target = self._resolve(relative_path)
|
|
if not target.exists():
|
|
raise FileNotFoundError(tr("gui_file.target_not_found"))
|
|
|
|
parent = target.parent
|
|
try:
|
|
sanitized = self._sanitize_entry_name(new_name)
|
|
except ValueError:
|
|
raise ValueError(tr("gui_file.new_name_empty"))
|
|
new_path = parent / sanitized
|
|
if new_path.exists():
|
|
raise FileExistsError(tr("gui_file.target_name_exists"))
|
|
|
|
target.rename(new_path)
|
|
return self._to_relative(new_path)
|
|
|
|
def copy_entries(self, relative_paths: List[str], destination_relative: str) -> Dict[str, str]:
|
|
destination = self._resolve(destination_relative)
|
|
if not destination.exists() or not destination.is_dir():
|
|
raise NotADirectoryError(tr("gui_file.destination_not_directory"))
|
|
|
|
results: Dict[str, str] = {}
|
|
for rel in relative_paths:
|
|
source = self._resolve(rel)
|
|
if not source.exists():
|
|
results[rel] = "missing"
|
|
continue
|
|
try:
|
|
target = self._unique_name(destination, source.name)
|
|
if source.is_dir():
|
|
shutil.copytree(source, target)
|
|
else:
|
|
shutil.copy2(source, target)
|
|
results[rel] = self._to_relative(target)
|
|
except Exception as exc:
|
|
results[rel] = f"error: {exc}"
|
|
return results
|
|
|
|
def move_entries(self, relative_paths: List[str], destination_relative: str) -> Dict[str, str]:
|
|
destination = self._resolve(destination_relative)
|
|
if not destination.exists() or not destination.is_dir():
|
|
raise NotADirectoryError(tr("gui_file.destination_not_directory"))
|
|
|
|
results: Dict[str, str] = {}
|
|
for rel in relative_paths:
|
|
source = self._resolve(rel)
|
|
if not source.exists():
|
|
results[rel] = "missing"
|
|
continue
|
|
try:
|
|
target = destination / source.name
|
|
if target.exists():
|
|
target = self._unique_name(destination, source.name)
|
|
shutil.move(str(source), str(target))
|
|
results[rel] = self._to_relative(target)
|
|
except Exception as exc:
|
|
results[rel] = f"error: {exc}"
|
|
return results
|
|
|
|
# -------------------------
|
|
# 文本读写
|
|
# -------------------------
|
|
|
|
def read_text(self, relative_path: str) -> Tuple[str, str]:
|
|
target = self._resolve(relative_path)
|
|
if not target.exists():
|
|
raise FileNotFoundError(tr("gui_file.file_not_found"))
|
|
if not target.is_file():
|
|
raise IsADirectoryError(tr("gui_file.target_is_directory"))
|
|
|
|
size = target.stat().st_size
|
|
if size > self.MAX_TEXT_FILE_SIZE:
|
|
raise ValueError(tr("gui_file.file_too_large"))
|
|
|
|
try:
|
|
with open(target, "r", encoding="utf-8") as fh:
|
|
content = fh.read()
|
|
except UnicodeDecodeError as exc:
|
|
raise ValueError(tr("gui_file.not_utf8", error=exc)) from exc
|
|
|
|
return content, datetime.fromtimestamp(target.stat().st_mtime).isoformat()
|
|
|
|
def write_text(self, relative_path: str, content: str) -> Dict[str, str]:
|
|
target = self._resolve(relative_path)
|
|
if not target.exists():
|
|
raise FileNotFoundError(tr("gui_file.file_not_found"))
|
|
if not target.is_file():
|
|
raise IsADirectoryError(tr("gui_file.target_is_directory"))
|
|
|
|
size = len(content.encode("utf-8"))
|
|
if size > self.MAX_TEXT_FILE_SIZE:
|
|
raise ValueError(tr("gui_file.content_too_large"))
|
|
|
|
with open(target, "w", encoding="utf-8") as fh:
|
|
fh.write(content)
|
|
|
|
stat = target.stat()
|
|
return {
|
|
"path": self._to_relative(target),
|
|
"size": str(stat.st_size),
|
|
"modified_at": datetime.fromtimestamp(stat.st_mtime).isoformat(),
|
|
}
|
|
|
|
# -------------------------
|
|
# 上传与下载
|
|
# -------------------------
|
|
|
|
def prepare_upload(self, destination_relative: Optional[str], filename: str) -> Path:
|
|
destination = self._resolve(destination_relative)
|
|
if not destination.exists():
|
|
destination.mkdir(parents=True, exist_ok=True)
|
|
if not destination.is_dir():
|
|
raise NotADirectoryError(tr("gui_file.upload_target_not_directory"))
|
|
sanitized = filename.strip()
|
|
if not sanitized:
|
|
raise ValueError(tr("gui_file.filename_empty"))
|
|
target = destination / sanitized
|
|
target = self._unique_name(destination, target.name) if target.exists() else target
|
|
return target
|
|
|
|
def prepare_download(self, relative_path: str) -> Path:
|
|
target = self._resolve(relative_path)
|
|
if not target.exists():
|
|
raise FileNotFoundError(tr("gui_file.file_not_found"))
|
|
return target
|
|
|