fix(security): 第二轮全量 API 审计修复（角色穿越/宿主 GUI/限流收口）

背景：对全部 206 个 HTTP 路由 + SocketIO 事件逐个排查（主审 + 2 个子
智能体初筛后复核），另按用户要求专项审计前端渲染面（无新发现）。
完整报告见 _experiments/security_audit_2026-09-02/REPORT.md 第八章。

新漏洞修复：
- 多智能体角色 role_id 路径穿越（高）：save_custom_role 直接拼接
  "{role_id}.md"，POST body 的 role_id 无任何校验，可 ../ 穿越覆盖
  其他用户角色文件（跨用户提示词注入），本地已复现写入成功。
  修复：新增 validate_role_id 白名单（^[a-z0-9][a-z0-9_-]{0,63}$），
  role_store 存储层 + multi_agent API 层（POST/PUT/DELETE）双保险
- open-in-file-manager 漏 _is_host_mode_request 检查（同文件其余三个
  端点都有）：docker 模式下任意登录用户可触发宿主 GUI 弹窗+探测路径。
  修复：补 host 检查，docker 下 403
- /api/admin/secondary/verify 无限流且 CSRF 豁免：admin 会话前提下
  可在线爆破二级密码（纵深防御缺失）。修复：5 次/300s 用户维度限流
- WS client_chunk_log / client_stream_debug_log 无连接认证检查且
  无限流（日志写盘放大面）。修复：必须已认证连接 + 30 次/60s 滑窗
  + 桶表万级上限回收
- _get_conversation_file_path 直接拼 "{id}.json"（conv_ 前缀恰好阻碍
  直接穿越，属防御深度缺失）。修复：^conv_[A-Za-z0-9_-]+$ 白名单，
  已核验全部 5 个调用点与 temp_ 前缀排除路径不受影响
- /api/conversations/media/<id> 采信 entry mime_type 并 inline 返回。
  修复：text/html、image/svg+xml、xhtml 强制 octet-stream + attachment
  （与 file/content 的 SVG 策略对齐，防存储型 XSS 一致性收口）
- /api/status 每次心跳返回宿主绝对 project_path（泄露系统用户名与
  目录布局）。修复：docker 模式脱敏为容器视角 /workspace，host 不变

LLM 成本攻击止血（第一轮发现 5 的端点级落地）：
- POST /api/tasks（GUI 发消息主入口）：30 次/60s/user
- POST /api/conversations/<id>/compress（调 api_client.chat 做摘要）：
  5 次/300s/user
- 注：对话回顾 review 端点实为纯本地 Markdown 生成（不调 LLM），
  其「发送给模型」模式走 /api/tasks，已被上述限流覆盖；
  按 token 计费的完整配额方案仍遗留待产品决策

验证：全部 py_compile 通过；冒烟测试 6/6；validate_role_id 8 个恶意
样本全拒 + 3 个合法样本放行 + 穿越写入拦截回归通过；conv_id 白名单
4 组样本符合预期。端点级行为待服务重启后实测。

Co-authored-by: Astrion powered by Kimi-K3 <astrion-agent@users.noreply.github.com>

.env.example

@@ -44,6 +44,10 @@ WEB_SERVER_PORT=8091
 WEB_SERVER_HOST=0.0.0.0
 # 调试模式（同时控制 Flask reloader），生产保持 0
 WEB_SERVER_DEBUG=0
+# 可信反向代理列表（逗号分隔 IP，默认空=不信任任何 X-Forwarded-For）
+# 仅当部署在 nginx/CF 等反代之后时才需要配置，例如 ASTRION_TRUSTED_PROXIES=127.0.0.1
+# 直连部署切勿配置，否则登录限流可被伪造的 XFF 绕过
+# ASTRION_TRUSTED_PROXIES=
 
 # === 运行环境二进制（便携 release 包用，开发环境留默认即可）==================
 # 子智能体使用的 node 可执行文件；留 "node" 表示走系统 PATH
@@ -64,7 +68,11 @@ TERMINAL_SANDBOX_SHELL=/bin/bash
 TERMINAL_SANDBOX_NETWORK=bridge
 TERMINAL_SANDBOX_CPUS=0.5
 TERMINAL_SANDBOX_MEMORY=1g
+# 容器内进程数上限（防 fork bomb）
+TERMINAL_SANDBOX_PIDS_LIMIT=512
 # 附加绑定目录（逗号分隔，可留空）
+# 安全红线：严禁绑定 /var/run/docker.sock、/、/etc、/home、/root、~/.ssh、
+# 数据根目录或源码目录——绑定 docker.sock 等于把宿主机 root 交给所有用户
 TERMINAL_SANDBOX_BINDS=
 # 运行时路径及容器名称前缀
 TERMINAL_SANDBOX_BIN=docker
@@ -81,6 +89,10 @@ HOST_SANDBOX_MACOS_WRITABLE_PATHS=
 # === 资源控制 ================================================================
 PROJECT_MAX_STORAGE_MB=2048
 MAX_ACTIVE_USER_CONTAINERS=8
+# 单个用户最多同时保活的容器数（防单用户占满全局容器池）
+MAX_ACTIVE_CONTAINERS_PER_USER=3
+# 单条消息最大字符数（防超大消息撑爆模型上下文/存储）
+MAX_MESSAGE_CHARS=200000
 
 # === MCP 工具扩展（可选） =====================================================
 # 1=启用，0=禁用

config/limits.py

@@ -13,6 +13,9 @@ CODE_EXECUTION_TIMEOUT = 60
 TERMINAL_COMMAND_TIMEOUT = 120
 SEARCH_MAX_RESULTS = 10
 
+# 单条用户消息最大字符数（防成本攻击与内存放大；2026-09-02 安全审计新增）
+MAX_MESSAGE_CHARS = int(os.environ.get("MAX_MESSAGE_CHARS", "200000") or 200000)
+
 # 自动修复与工具调用限制（None 表示不限制）
 AUTO_FIX_TOOL_CALL = False
 AUTO_FIX_MAX_ATTEMPTS = 3

config/terminal.py

@@ -39,8 +39,11 @@ TERMINAL_SANDBOX_IMAGE = os.environ.get("TERMINAL_SANDBOX_IMAGE", "python:3.11-s
 TERMINAL_SANDBOX_MOUNT_PATH = os.environ.get("TERMINAL_SANDBOX_MOUNT_PATH", "/workspace")
 TERMINAL_SANDBOX_SHELL = os.environ.get("TERMINAL_SANDBOX_SHELL", "/bin/bash")
 TERMINAL_SANDBOX_NETWORK = os.environ.get("TERMINAL_SANDBOX_NETWORK", "bridge")
-TERMINAL_SANDBOX_CPUS = os.environ.get("TERMINAL_SANDBOX_CPUS", "")
-TERMINAL_SANDBOX_MEMORY = os.environ.get("TERMINAL_SANDBOX_MEMORY", "")
+# 安全默认：限制单容器 CPU/内存/PID，防止单用户资源耗尽型 DoS（2026-09-02 审计）。
+# 如需放开可显式设为空字符串（不推荐多用户部署放开）。
+TERMINAL_SANDBOX_CPUS = os.environ.get("TERMINAL_SANDBOX_CPUS", "1")
+TERMINAL_SANDBOX_MEMORY = os.environ.get("TERMINAL_SANDBOX_MEMORY", "1g")
+TERMINAL_SANDBOX_PIDS_LIMIT = os.environ.get("TERMINAL_SANDBOX_PIDS_LIMIT", "512")
 TERMINAL_SANDBOX_BINDS = _parse_bindings(os.environ.get("TERMINAL_SANDBOX_BINDS", ""))
 TERMINAL_SANDBOX_BIN = os.environ.get("TERMINAL_SANDBOX_BIN", "docker")
 TERMINAL_SANDBOX_NAME_PREFIX = os.environ.get("TERMINAL_SANDBOX_NAME_PREFIX", "agent-term")
@@ -53,6 +56,8 @@ TERMINAL_SANDBOX_REQUIRE = os.environ.get("TERMINAL_SANDBOX_REQUIRE", "0") not i
 LINUX_SAFETY = os.environ.get("LINUX_SAFETY", "0") not in {"0", "false", "False"}
 TOOLBOX_TERMINAL_IDLE_SECONDS = int(os.environ.get("TOOLBOX_TERMINAL_IDLE_SECONDS", "900"))
 MAX_ACTIVE_USER_CONTAINERS = int(os.environ.get("MAX_ACTIVE_USER_CONTAINERS", "8"))
+# 每用户同时活跃的容器上限（防单用户多工作区占满全局容器池，2026-09-02 审计新增）
+MAX_ACTIVE_CONTAINERS_PER_USER = int(os.environ.get("MAX_ACTIVE_CONTAINERS_PER_USER", "3"))
 HOST_EXECUTION_MODE_DEFAULT = os.environ.get("HOST_EXECUTION_MODE_DEFAULT", "sandbox").strip().lower()
 # 沙箱可写路径的「部署通道」（逗号分隔）。路径授权只有两个来源：
 # config/host_sandbox_policy.json（前端「路径授权」UI）+ 本变量（真·环境变量），
@@ -83,6 +88,7 @@ __all__ = [
     "TERMINAL_SANDBOX_NETWORK",
     "TERMINAL_SANDBOX_CPUS",
     "TERMINAL_SANDBOX_MEMORY",
+    "TERMINAL_SANDBOX_PIDS_LIMIT",
     "TERMINAL_SANDBOX_BINDS",
     "TERMINAL_SANDBOX_BIN",
     "TERMINAL_SANDBOX_NAME_PREFIX",
@@ -91,6 +97,7 @@ __all__ = [
     "LINUX_SAFETY",
     "TOOLBOX_TERMINAL_IDLE_SECONDS",
     "MAX_ACTIVE_USER_CONTAINERS",
+    "MAX_ACTIVE_CONTAINERS_PER_USER",
     "HOST_EXECUTION_MODE_DEFAULT",
     "HOST_SANDBOX_MACOS_WRITABLE_PATHS",
     "HOST_SANDBOX_NETWORK_PERMISSION",

modules/api_user_manager.py

@@ -115,6 +115,10 @@ class ApiUserManager:
         ws_id = (workspace_id or "default").strip()
         if not ws_id:
             ws_id = "default"
+        # 安全：workspace_id 参与路径拼接，必须严格白名单（防 `..`/斜杠穿越）
+        import re as _re
+        if not _re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,39}", ws_id) or ".." in ws_id:
+            raise ValueError(tr("api_user_mgr.invalid_workspace_id"))
 
         user_root = (self.workspace_root / username).resolve()
         shared_dir = user_root / "shared"
@@ -284,7 +288,15 @@ class ApiUserManager:
         ws_id = (workspace_id or "").strip()
         if not ws_id:
             return False
+        # 安全：同 ensure_workspace 的严格白名单校验，防路径穿越删除任意目录
+        import re as _re
+        if not _re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,39}", ws_id) or ".." in ws_id:
+            return False
         work_root = (self.workspace_root / username / "workspaces" / ws_id).resolve()
+        # 二次防护：解析后的落点必须严格位于该用户的 workspaces 目录内
+        expected_parent = (self.workspace_root / username / "workspaces").resolve()
+        if work_root.parent != expected_parent:
+            return False
         if not work_root.exists():
             return False
         import shutil

modules/container_file_proxy.py

@@ -19,7 +19,8 @@ import shutil
 def _resolve(root: pathlib.Path, rel: str) -> pathlib.Path:
     base = root.resolve()
     target = (base / rel).resolve()
-    if not str(target).startswith(str(base)):
+    # 带分隔符边界的包含判断，防 startswith 前缀碰撞（如 /workspace-evil）
+    if target != base and not str(target).startswith(str(base) + __import__('os').sep):
         raise ValueError("路径越界: %s" % rel)
     return target
 
@@ -82,7 +83,12 @@ def _create_folder(root, payload):
 
 def _delete_folder(root, payload):
     rel = payload.get("path")
+    # 安全：禁止删除工作区根本身（防 rmtree 抹掉整个 bind-mount 工作区）
+    if not rel or str(rel).strip() in {"", ".", "./", ".."}:
+        return {"success": False, "error": "不允许删除工作区根目录"}
     target = _resolve(root, rel)
+    if target == root.resolve():
+        return {"success": False, "error": "不允许删除工作区根目录"}
     if not target.exists():
         return {"success": False, "error": "文件夹不存在"}
     if not target.is_dir():

modules/file_manager/crud_mixin.py

@@ -210,6 +210,14 @@ class CrudMixin:
         if not valid:
             return {"success": False, "error": error}
 
+        # 安全：禁止删除工作区根目录本身（空路径/点路径会解析为根；
+        # docker 模式下该目录是宿主 bind mount，rmtree 会真实删除宿主文件）
+        try:
+            if full_path.resolve() == Path(self.project_path).resolve():
+                return {"success": False, "error": tr("file_manager.cannot_delete_workspace_root")}
+        except Exception:
+            pass
+
         if not full_path.exists():
             return {"success": False, "error": tr("file_manager.folder_not_found")}
         

modules/gui_file_manager.py

@@ -167,6 +167,15 @@ class GuiFileManager:
     # 基本操作
     # -------------------------
 
+    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():
@@ -174,9 +183,7 @@ class GuiFileManager:
         if not parent.is_dir():
             raise NotADirectoryError(tr("gui_file.parent_not_directory"))
 
-        sanitized = name.strip()
-        if not sanitized:
-            raise ValueError(tr("gui_file.name_empty"))
+        sanitized = self._sanitize_entry_name(name)
 
         target = parent / sanitized
         if target.exists():
@@ -215,8 +222,9 @@ class GuiFileManager:
             raise FileNotFoundError(tr("gui_file.target_not_found"))
 
         parent = target.parent
-        sanitized = new_name.strip()
-        if not sanitized:
+        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():

modules/i18n_messages/api_conversation.py

@@ -7,6 +7,11 @@ Keys are prefixed with ``conversation.`` (the core table already owns
 """
 
 MESSAGES = {
+    # ── _get_conversation_file_path（对话 ID 白名单校验） ──
+    "conversation.invalid_conversation_id": {
+        "zh-CN": "对话 ID 不合法",
+        "en-US": "Invalid conversation ID",
+    },
     # ── _build_safe_load_result（安全导航加载结果，返还前端） ──
     "conversation.load_failed_not_found": {
         "zh-CN": "对话不存在或加载失败",

modules/i18n_messages/api_multi_agent.py

@@ -30,6 +30,10 @@ MESSAGES = {
         "zh-CN": "role_id/name/body_prompt 必填",
         "en-US": "role_id/name/body_prompt are required",
     },
+    "multi_agent_api.role_id_invalid": {
+        "zh-CN": "角色 ID 不合法：{role_id}（仅允许小写字母/数字开头，后接小写字母/数字/连字符/下划线）",
+        "en-US": "Invalid role ID: {role_id} (must start with lowercase letter/digit, followed by lowercase letters/digits/hyphens/underscores)",
+    },
     "multi_agent_api.cannot_override_preset_role": {
         "zh-CN": "不能覆盖预设角色 {role_id}",
         "en-US": "Cannot override preset role {role_id}",

modules/i18n_messages/api_tasks.py

@@ -16,6 +16,10 @@ MESSAGES = {
         "zh-CN": "消息不能为空",
         "en-US": "Message cannot be empty",
     },
+    "tasks.message_too_long": {
+        "zh-CN": "消息内容过长",
+        "en-US": "Message is too long",
+    },
     "tasks.workspace_unavailable": {
         "zh-CN": "工作区不可用",
         "en-US": "Workspace is unavailable",

modules/i18n_messages/api_v1.py

@@ -111,4 +111,16 @@ MESSAGES = {
         "zh-CN": "保存失败: {error}",
         "en-US": "Failed to save: {error}",
     },
+    "api_v1.invalid_resource_name": {
+        "zh-CN": "名称只能包含字母、数字、下划线和连字符（最长64字符）",
+        "en-US": "Name may only contain letters, digits, underscores and hyphens (max 64 chars)",
+    },
+    "api_v1.folder_download_removed": {
+        "zh-CN": "文件夹打包下载功能已下线",
+        "en-US": "Folder archive download has been removed",
+    },
+    "api_v1.message_too_long": {
+        "zh-CN": "消息内容过长",
+        "en-US": "Message is too long",
+    },
 }

modules/i18n_messages/file_manager.py

@@ -154,6 +154,10 @@ MESSAGES = {
         "zh-CN": "文件夹不存在",
         "en-US": "Folder not found",
     },
+    "file_manager.cannot_delete_workspace_root": {
+        "zh-CN": "不允许删除工作区根目录",
+        "en-US": "Deleting the workspace root is not allowed",
+    },
     "file_manager.not_a_folder": {
         "zh-CN": "不是文件夹",
         "en-US": "Not a folder",

modules/i18n_messages/infra.py

@@ -219,6 +219,10 @@ MESSAGES = {
         "zh-CN": "资源繁忙：容器配额已用尽，请稍候再试。",
         "en-US": "Busy: container quota exhausted, please try again later",
     },
+    "container_mgr.per_user_quota_exhausted": {
+        "zh-CN": "资源繁忙：您的容器数量已达上限（{limit} 个），请先释放其他工作区的容器。",
+        "en-US": "Busy: you have reached your container limit ({limit}); please release containers of other workspaces first",
+    },
     "container_mgr.runtime_not_found": {
         "zh-CN": "未找到容器运行时 {runtime}",
         "en-US": "Container runtime not found: {runtime}",

modules/i18n_messages/managers.py

@@ -140,6 +140,10 @@ MESSAGES = {
         "zh-CN": "用户名不能为空",
         "en-US": "Username cannot be empty",
     },
+    "api_user_mgr.invalid_workspace_id": {
+        "zh-CN": "workspace_id 只能包含字母、数字、点、下划线或连字符（1-40 位，以字母或数字开头）",
+        "en-US": "workspace_id may only contain letters, digits, dots, underscores or hyphens (1-40 chars, starting with a letter or digit)",
+    },
     "api_user_mgr.users_file_parse_failed": {
         "zh-CN": "无法解析 API 用户文件: {file_path} ({error})",
         "en-US": "Failed to parse API user file: {file_path} ({error})",

modules/i18n_messages/server_core_leftovers.py

@@ -34,6 +34,10 @@ MESSAGES = {
         "zh-CN": "缺少 path",
         "en-US": "Missing path",
     },
+    "files.folder_download_removed": {
+        "zh-CN": "文件夹打包下载功能已下线",
+        "en-US": "Folder archive download has been removed",
+    },
     "files.missing_path_or_content": {
         "zh-CN": "缺少 path 或 content",
         "en-US": "Missing path or content",

modules/multi_agent/role_store.py

@@ -33,8 +33,22 @@ except ImportError:
     CUSTOM_ROLES_DIR = ""
     WEB_PRESET_ROLES_DIR = ""
 
+from modules.i18n import tr
+
 FRONTMATTER_RE = re.compile(r"^---\s*\n(?P<body>.*?)\n---\s*\n(?P<rest>.*)$", re.S)
 
+# 角色 ID 白名单：小写字母数字开头，允许连字符/下划线（与现有 ui-operator 等预设兼容）。
+# role_id 直接参与文件路径拼接（{role_id}.md），必须拒绝 /、.. 等路径字符。
+_ROLE_ID_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$")
+
+
+def validate_role_id(role_id: str) -> str:
+    """校验并返回合法 role_id；非法时 raise ValueError（防路径穿越写入）。"""
+    cleaned = (role_id or "").strip()
+    if not _ROLE_ID_RE.match(cleaned):
+        raise ValueError(tr("multi_agent_api.role_id_invalid", role_id=repr(cleaned)))
+    return cleaned
+
 
 # ----- 目录推断 -----
 
@@ -363,6 +377,7 @@ def save_custom_role(
     custom_dir: Optional[str | Path] = None,
 ) -> Path:
     """保存角色到自定义目录。不传 custom_dir 时回退到 host 运行态目录。"""
+    validate_role_id(role.role_id)
     if custom_dir:
         c_dir = Path(custom_dir).expanduser().resolve()
     else:
@@ -378,6 +393,7 @@ def delete_custom_role(
     custom_dir: Optional[str | Path] = None,
 ) -> bool:
     """删除用户自定义角色。不传 custom_dir 时回退到 host 运行态目录。"""
+    role_id = validate_role_id(role_id)
     if custom_dir:
         c_dir = Path(custom_dir).expanduser().resolve()
     else:

modules/user_container_manager.py

@@ -15,6 +15,7 @@ from pathlib import Path
 from typing import Dict, Optional
 
 from config import (
+    MAX_ACTIVE_CONTAINERS_PER_USER,
     MAX_ACTIVE_USER_CONTAINERS,
     OUTPUT_FORMATS,
     TERMINAL_SANDBOX_BIN,
@@ -27,6 +28,7 @@ from config import (
     TERMINAL_SANDBOX_MOUNT_PATH,
     TERMINAL_SANDBOX_NAME_PREFIX,
     TERMINAL_SANDBOX_NETWORK,
+    TERMINAL_SANDBOX_PIDS_LIMIT,
     TERMINAL_SANDBOX_REQUIRE,
     LOGS_DIR,
     LINUX_SAFETY,
@@ -80,6 +82,7 @@ class UserContainerManager:
         self.network = TERMINAL_SANDBOX_NETWORK
         self.cpus = TERMINAL_SANDBOX_CPUS
         self.memory = TERMINAL_SANDBOX_MEMORY
+        self.pids_limit = TERMINAL_SANDBOX_PIDS_LIMIT
         self.binds = list(TERMINAL_SANDBOX_BINDS)
         self.sandbox_bin = TERMINAL_SANDBOX_BIN or "docker"
         self.name_prefix = TERMINAL_SANDBOX_NAME_PREFIX or "agent-user"
@@ -158,6 +161,16 @@ class UserContainerManager:
             if not self._has_capacity(key):
                 raise RuntimeError(tr("container_mgr.quota_exhausted"))
 
+            # 每用户容器配额：防单用户多工作区占满全局容器池，挤占其他用户
+            per_user_limit = MAX_ACTIVE_CONTAINERS_PER_USER
+            if per_user_limit > 0:
+                owned = sum(
+                    1 for k in self._containers
+                    if k == username_norm or k.startswith(f"{username_norm}::")
+                )
+                if owned >= per_user_limit:
+                    raise RuntimeError(tr("container_mgr.per_user_quota_exhausted", limit=per_user_limit))
+
             # Important: create container using the cache key so each workspace gets its own container name.
             handle = self._create_handle(key, workspace, mode)
             self._containers[key] = handle
@@ -327,6 +340,12 @@ class UserContainerManager:
             cmd += ["--cpus", str(self.cpus)]
         if self.memory:
             cmd += ["--memory", str(self.memory)]
+            # 锁定 swap 用量等于内存上限，防止 swap 绕过内存限制
+            cmd += ["--memory-swap", str(self.memory)]
+        # 安全默认（2026-09-02 审计）：PID 上限防 fork bomb；禁提权防 setuid 类攻击。
+        if self.pids_limit:
+            cmd += ["--pids-limit", str(self.pids_limit)]
+        cmd += ["--security-opt", "no-new-privileges:true"]
         for bind in self.binds:
             chunk = bind.strip()
             if chunk:

server/admin.py

@@ -117,6 +117,7 @@ def admin_secondary_status():
 @admin_bp.route('/api/admin/secondary/verify', methods=['POST'])
 @login_required
 @admin_required
+@rate_limited("admin_secondary_verify", 5, 300, scope="user")
 def admin_secondary_verify():
     payload = request.get_json() or {}
     password = str(payload.get("password") or "").strip()

server/api_v1.py

@@ -2,14 +2,13 @@
 from __future__ import annotations
 import os
 import json
-import zipfile
-from io import BytesIO
 from pathlib import Path
 from typing import Dict, Any, Optional
 
 from flask import Blueprint, request, jsonify, send_file, session
 
 from .api_auth import api_token_required
+from .security import rate_limited
 from .tasks import task_manager
 from .context import get_user_resources, ensure_conversation_loaded, get_upload_guard, apply_conversation_overrides
 from .utils_common import sanitize_filename_preserve_unicode
@@ -69,6 +68,7 @@ def list_workspaces_api():
 
 @api_v1_bp.route("/workspaces", methods=["POST"])
 @api_token_required
+@rate_limited("api_v1_create_ws", 10, 300, scope="user")
 def create_workspace_api():
     username = session.get("username")
     payload = request.get_json(silent=True) or {}
@@ -120,6 +120,25 @@ def delete_workspace_api(workspace_id: str):
     return jsonify({"success": True, "workspace_id": ws_id})
 
 
+def _path_within(base: Path, target: Path) -> bool:
+    """严格的路径包含判断（带分隔符边界），避免 startswith 前缀碰撞。"""
+    if target == base:
+        return True
+    return str(target).startswith(str(base) + os.sep)
+
+
+def _validate_resource_name(name: str) -> str:
+    """校验 prompts/personalizations 等资源名，防路径穿越写入。
+
+    仅允许字母数字/下划线/连字符，最长 64；不含点号（防 `..` 与后缀伪装）。
+    """
+    import re
+    candidate = (name or "").strip()
+    if not candidate or not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", candidate):
+        raise ValueError(tr("api_v1.invalid_resource_name"))
+    return candidate
+
+
 def _within_uploads(workspace, rel_path: str) -> Path:
     base = Path(workspace.uploads_dir).resolve()
     rel = rel_path or ""
@@ -128,7 +147,7 @@ def _within_uploads(workspace, rel_path: str) -> Path:
         rel = rel.split("user_upload/", 1)[1]
     rel = rel.lstrip("/").strip()
     target = (base / rel).resolve()
-    if not str(target).startswith(str(base)):
+    if not _path_within(base, target):
         raise ValueError(tr("api_v1_raise.invalid_path"))
     return target
 
@@ -151,7 +170,7 @@ def _within_project(workspace, rel_path: str, default_to_project_root: bool = Tr
         rel = rel.split("/workspace/", 1)[1]
     rel = rel.lstrip("/")
     target = (base / rel).resolve()
-    if not str(target).startswith(str(base)):
+    if not _path_within(base, target):
         raise ValueError(tr("api_v1_raise.invalid_path"))
     return target
 
@@ -207,6 +226,7 @@ def _resolve_workspace(username: str, workspace_id: str):
 
 @api_v1_bp.route("/workspaces/<workspace_id>/conversations", methods=["POST"])
 @api_token_required
+@rate_limited("api_v1_create_conv", 30, 60, scope="user")
 def create_conversation_api(workspace_id: str):
     username = session.get("username")
     ws = _resolve_workspace(username, workspace_id)
@@ -233,11 +253,15 @@ def create_conversation_api(workspace_id: str):
 
 @api_v1_bp.route("/workspaces/<workspace_id>/messages", methods=["POST"])
 @api_token_required
+@rate_limited("api_v1_send_msg", 20, 60, scope="user")
 def send_message_api(workspace_id: str):
     username = session.get("username")
     ws = _resolve_workspace(username, workspace_id)
     payload = request.get_json() or {}
     message = (payload.get("message") or "").strip()
+    from config import MAX_MESSAGE_CHARS
+    if len(message) > MAX_MESSAGE_CHARS:
+        return jsonify({"success": False, "error": tr("api_v1.message_too_long")}), 400
     images = payload.get("images") or []
     conversation_id = payload.get("conversation_id")
     model_key = payload.get("model_key")
@@ -260,11 +284,13 @@ def send_message_api(workspace_id: str):
     # 应用自定义 prompt/personalization（如果提供）
     try:
         if prompt_name:
+            prompt_name = _validate_resource_name(prompt_name)
             prompt_path = _prompt_dir(workspace) / f"{prompt_name}.txt"
             if not prompt_path.exists():
                 return jsonify({"success": False, "error": tr("api_v1.prompt_not_found")}), 404
             terminal.context_manager.custom_system_prompt = prompt_path.read_text(encoding="utf-8")
         if personalization_name:
+            personalization_name = _validate_resource_name(personalization_name)
             pers_path = _personalization_dir(workspace) / f"{personalization_name}.json"
             if not pers_path.exists():
                 return jsonify({"success": False, "error": tr("api_v1.personalization_not_found")}), 404
@@ -427,6 +453,7 @@ def cancel_task_api_v1(task_id: str):
 
 @api_v1_bp.route("/workspaces/<workspace_id>/files/upload", methods=["POST"])
 @api_token_required
+@rate_limited("api_v1_upload", 20, 300, scope="user")
 def upload_file_api(workspace_id: str):
     username = session.get("username")
     ws = _resolve_workspace(username, workspace_id)
@@ -538,16 +565,9 @@ def download_file_api(workspace_id: str):
         return jsonify({"success": False, "error": tr("api_v1.file_does_not_exist")}), 404
 
     if target.is_dir():
-        memory_file = BytesIO()
-        with zipfile.ZipFile(memory_file, mode='w', compression=zipfile.ZIP_DEFLATED) as zf:
-            for root, _, files in os.walk(target):
-                for file in files:
-                    full_path = Path(root) / file
-                    arcname = full_path.relative_to(workspace.project_path)
-                    zf.write(full_path, arcname=str(arcname))
-        memory_file.seek(0)
-        download_name = f"{target.name}.zip"
-        return send_file(memory_file, as_attachment=True, download_name=download_name, mimetype='application/zip')
+        # 文件夹打包下载已下线（2026-09-02 安全审计：os.walk/zipfile 跟随文件型
+        # 符号链接，可泄露宿主任意文件）
+        return jsonify({"success": False, "error": tr("api_v1.folder_download_removed")}), 410
     return send_file(target, as_attachment=True, download_name=target.name)
 
 
@@ -580,6 +600,10 @@ def get_prompt_api(name: str):
     _, workspace = get_user_resources(username, workspace_id=ws.workspace_id)
     if not workspace:
         return jsonify({"success": False, "error": tr("api_v1.system_not_initialized")}), 503
+    try:
+        name = _validate_resource_name(name)
+    except ValueError as exc:
+        return jsonify({"success": False, "error": str(exc)}), 400
     p = _prompt_dir(workspace) / f"{name}.txt"
     if not p.exists():
         return jsonify({"success": False, "error": tr("api_v1.prompt_not_found")}), 404
@@ -599,6 +623,10 @@ def create_prompt_api():
     content = payload.get("content") or ""
     if not name:
         return jsonify({"success": False, "error": tr("api_v1.name_empty")}), 400
+    try:
+        name = _validate_resource_name(name)
+    except ValueError as exc:
+        return jsonify({"success": False, "error": str(exc)}), 400
     p = _prompt_dir(workspace) / f"{name}.txt"
     p.write_text(content, encoding="utf-8")
     return jsonify({"success": True, "name": name})
@@ -632,6 +660,10 @@ def get_personalization_api(name: str):
     _, workspace = get_user_resources(username, workspace_id=ws.workspace_id)
     if not workspace:
         return jsonify({"success": False, "error": tr("api_v1.system_not_initialized")}), 503
+    try:
+        name = _validate_resource_name(name)
+    except ValueError as exc:
+        return jsonify({"success": False, "error": str(exc)}), 400
     p = _personalization_dir(workspace) / f"{name}.json"
     if not p.exists():
         return jsonify({"success": False, "error": tr("api_v1.personalization_not_found")}), 404
@@ -661,6 +693,10 @@ def create_personalization_api():
         return jsonify({"success": False, "error": tr("api_v1.name_empty")}), 400
     if content is None:
         return jsonify({"success": False, "error": tr("api_v1.content_empty")}), 400
+    try:
+        name = _validate_resource_name(name)
+    except ValueError as exc:
+        return jsonify({"success": False, "error": str(exc)}), 400
     p = _personalization_dir(workspace) / f"{name}.json"
     try:
         if not isinstance(content, dict):

server/app_legacy.py

@@ -1142,9 +1142,16 @@ def resource_busy_page():
 
 @app.route('/api/client_debug_log', methods=['POST'])
 def client_debug_log():
-    """接收前端目标模式等调试日志"""
+    """接收前端目标模式等调试日志（限流 + 长度截断，防止未认证磁盘攻击）。"""
+    from server.security import check_rate_limit, get_client_ip
+    limited, retry_after = check_rate_limit("client_debug_log", 10, 60, get_client_ip())
+    if limited:
+        return jsonify({'ok': False, 'error': 'rate_limited', 'retry_after': retry_after}), 429
     try:
         data = request.get_json(silent=True) or {}
+        serialized = json.dumps(data, ensure_ascii=False)
+        if len(serialized) > 4000:  # 单条硬截断，避免大 payload 灌入日志
+            data = {"truncated": True, "raw_preview": serialized[:4000]}
         entry = dict(data)
         entry.setdefault('server_ts', time.time())
         log_goal_mode_debug_entry(entry)

server/auth.py

@@ -23,6 +23,7 @@ from .security import (
     register_failure,
     is_action_blocked,
     clear_failures,
+    get_client_ip,
 )
 from . import state
 from .utils_common import debug_log
@@ -37,8 +38,9 @@ def auth_debug_log(message: str):
     ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
     line = f"[{ts}] {message}"
     try:
-        with AUTH_DEBUG_FILE.open("a", encoding="utf-8") as f:
-            f.write(line + "\n")
+        # 走统一轮转，防止未认证请求刷爆磁盘（此前为无上限 append）
+        from utils.log_rotation import append_line
+        append_line(AUTH_DEBUG_FILE, line)
     except Exception:
         pass
     try:
@@ -139,7 +141,9 @@ def login():
     data = request.get_json() or {}
     email = (data.get('email') or '').strip()
     password = data.get('password') or ''
-    client_ip = request.headers.get('X-Forwarded-For', '').split(',')[0].strip() or request.remote_addr or 'unknown'
+    # XFF 仅在可信代理后采信（见 security.get_client_ip），直连时以真实对端为准
+    client_ip = get_client_ip()
+    account_key = (email or '').strip().lower() or 'unknown'
 
     limited, retry_after = check_rate_limit("login", 10, 60, client_ip)
     if limited:
@@ -149,8 +153,14 @@ def login():
     if blocked:
         return jsonify({"success": False, "error": tr("auth.too_many_attempts", seconds=block_for), "retry_after": block_for}), 429
 
+    # 账号维度锁定：防止伪造/轮换来源 IP 对同一账号持续爆破
+    blocked, block_for = is_action_blocked("login_account", identifier=account_key)
+    if blocked:
+        return jsonify({"success": False, "error": tr("auth.too_many_attempts", seconds=block_for), "retry_after": block_for}), 429
+
     record = state.user_manager.authenticate(email, password)
     if not record:
+        register_failure("login_account", state.FAILED_LOGIN_LIMIT, state.FAILED_LOGIN_LOCK_SECONDS, identifier=account_key)
         wait_seconds = register_failure("login", state.FAILED_LOGIN_LIMIT, state.FAILED_LOGIN_LOCK_SECONDS, identifier=client_ip)
         error_payload = {"success": False, "error": tr("auth.invalid_credentials")}
         status_code = 401
@@ -193,6 +203,7 @@ def login():
     session.permanent = True
     _issue_login_nonce(record.username)
     clear_failures("login", identifier=client_ip)
+    clear_failures("login_account", identifier=account_key)
     try:
         state.container_manager.ensure_container(
             record.username,
@@ -211,8 +222,15 @@ def login():
 
 @auth_bp.route('/host-login', methods=['POST'])
 def host_login():
-    """宿主机模式一键进入（仅当 TERMINAL_SANDBOX_MODE=host 时可用）。"""
-    if (TERMINAL_SANDBOX_MODE or "").lower() != "host":
+    """宿主机模式一键进入（仅当 TERMINAL_SANDBOX_MODE=host 时可用）。
+
+    该入口无任何凭证，因此只允许本机回环地址直连调用（host 模式定位为本机单人使用）；
+    经反向代理/远程访问时一律拒绝，防止公网部署下任何人一键获得 admin 会话。
+    """
+    if (TERMINAL_SANDBOX_MODE or "").lower() != "host" or LINUX_SAFETY:
+        return jsonify({"success": False, "error": tr("auth.host_mode_disabled")}), 403
+    remote_addr = (request.remote_addr or "").strip()
+    if remote_addr not in {"127.0.0.1", "::1", "localhost"}:
         return jsonify({"success": False, "error": tr("auth.host_mode_disabled")}), 403
     if not state.container_manager.has_capacity("host"):
         return jsonify({"success": False, "error": tr("auth.resource_busy")}), 503
@@ -293,14 +311,22 @@ def register():
     invite_code = (data.get('invite_code') or '').strip()
 
     from .security import get_client_ip
-    limited, retry_after = check_rate_limit("register", 5, 300, get_client_ip())
+    client_ip = get_client_ip()
+    limited, retry_after = check_rate_limit("register", 5, 300, client_ip)
     if limited:
         return jsonify({"success": False, "error": tr("auth.register_rate_limited"), "retry_after": retry_after}), 429
+
+    # 邀请码爆破锁定：连续失败达到阈值后按来源锁定一段时间
+    blocked, block_for = is_action_blocked("register_invite", identifier=client_ip)
+    if blocked:
+        return jsonify({"success": False, "error": tr("auth.too_many_attempts", seconds=block_for), "retry_after": block_for}), 429
+
     try:
         state.user_manager.register_user(username, email, password, invite_code)
         auth_debug_log(f"[auth_debug] POST /register success username={username}")
         return jsonify({"success": True})
     except ValueError as exc:
+        register_failure("register_invite", state.FAILED_LOGIN_LIMIT, state.FAILED_LOGIN_LOCK_SECONDS, identifier=client_ip)
         return jsonify({"success": False, "error": str(exc)}), 400
     except Exception as exc:
         return jsonify({"success": False, "error": str(exc)}), 500

server/chat/files.py

@@ -4,8 +4,6 @@ import json, time
 from datetime import datetime
 from typing import Dict, Any, Optional
 from pathlib import Path
-from io import BytesIO
-import zipfile
 import os
 
 from flask import Blueprint, jsonify, request, session, send_file
@@ -252,42 +250,10 @@ def download_file_api(terminal: WebTerminal, workspace: UserWorkspace, username:
         download_name=full_path.name
     )
 
-@chat_bp.route('/api/download/folder')
-@api_login_required
-@with_terminal
-def download_folder_api(terminal: WebTerminal, workspace: UserWorkspace, username: str):
-    """打包并下载文件夹"""
-    path = (request.args.get('path') or '').strip()
-    if not path:
-        return jsonify({"success": False, "error": tr("chat_files.missing_path_param")}), 400
-
-    valid, error, full_path = terminal.file_manager._validate_path(path)
-    if not valid or full_path is None:
-        return jsonify({"success": False, "error": error or tr("chat_files.path_validation_failed")}), 400
-    if not full_path.exists() or not full_path.is_dir():
-        return jsonify({"success": False, "error": tr("chat_files.folder_not_found")}), 404
-
-    buffer = BytesIO()
-    folder_name = Path(path).name or full_path.name or "archive"
-
-    with zipfile.ZipFile(buffer, 'w', zipfile.ZIP_DEFLATED) as zip_buffer:
-        # 确保目录本身被包含
-        zip_buffer.write(full_path, arcname=folder_name + '/')
-
-        for item in full_path.rglob('*'):
-            relative_name = Path(folder_name) / item.relative_to(full_path)
-            if item.is_dir():
-                zip_buffer.write(item, arcname=str(relative_name) + '/')
-            else:
-                zip_buffer.write(item, arcname=str(relative_name))
-
-    buffer.seek(0)
-    return send_file(
-        buffer,
-        mimetype='application/zip',
-        as_attachment=True,
-        download_name=f"{folder_name}.zip"
-    )
+# 文件夹打包下载功能已彻底移除（2026-09-02 安全审计）：
+# 原 /api/download/folder 端点在打包时不对 rglob 子项做符号链接/越界校验，
+# 攻击者在容器工作区内创建指向宿主绝对路径的文件型符号链接即可经 zip 泄露宿主任意文件
+# （含 settings.json / .env / 其他用户数据）。前端已无调用，功能直接下线。
 
 
 # 文件预览允许 inline 展示的 MIME 前缀白名单
@@ -314,13 +280,16 @@ _FILE_CONTENT_INLINE_TEXT_EXTS = {
 
 def _resolve_inline_mime(full_path: Path) -> str:
     """猜测 inline 预览用的 MIME 类型，不在白名单内的退化为 octet-stream。"""
+    suffix = full_path.suffix.lower()
+    # SVG 虽为 image/*，但可携带脚本，inline 渲染即存储型 XSS——与 HTML 同策略强制下载
+    if suffix == '.svg':
+        return 'application/octet-stream'
     guessed, _ = mimetypes.guess_type(str(full_path))
     if guessed:
         if guessed.startswith(_FILE_CONTENT_INLINE_MIME_PREFIXES):
             return guessed
         return 'application/octet-stream'
     # mimetypes 未识别，按扩展名兑底
-    suffix = full_path.suffix.lower()
     if suffix in _FILE_CONTENT_INLINE_TEXT_EXTS:
         return 'text/plain; charset=utf-8'
     if suffix == '.pdf':

server/chat/misc.py

@@ -71,7 +71,7 @@ def get_monitor_snapshot_api():
     stage = (request.args.get('stage') or 'before').lower()
     if stage not in {'before', 'after'}:
         stage = 'before'
-    snapshot = get_cached_monitor_snapshot(execution_id, stage)
+    snapshot = get_cached_monitor_snapshot(execution_id, stage, username=get_current_username())
     if not snapshot:
         return jsonify({
             'success': False,

server/chat_flow_tool_loop.py

@@ -944,7 +944,7 @@ async def _execute_tool_calls_impl(*, web_terminal, tool_calls, sender, messages
             snapshot_path = resolve_monitor_path(arguments)
             monitor_snapshot = capture_monitor_snapshot(web_terminal.file_manager, snapshot_path, MONITOR_SNAPSHOT_CHAR_LIMIT, debug_log)
             if monitor_snapshot:
-                cache_monitor_snapshot(tool_display_id, 'before', monitor_snapshot)
+                cache_monitor_snapshot(tool_display_id, 'before', monitor_snapshot, username=username)
         elif function_name in MONITOR_MEMORY_TOOLS:
             memory_snapshot_type = (arguments.get('memory_type') or 'main').lower()
             before_entries = None
@@ -957,7 +957,7 @@ async def _execute_tool_calls_impl(*, web_terminal, tool_calls, sender, messages
                     'memory_type': memory_snapshot_type,
                     'entries': before_entries
                 }
-                cache_monitor_snapshot(tool_display_id, 'before', monitor_snapshot)
+                cache_monitor_snapshot(tool_display_id, 'before', monitor_snapshot, username=username)
 
         sender('tool_start', {
             'id': tool_display_id,
@@ -1300,7 +1300,7 @@ async def _execute_tool_calls_impl(*, web_terminal, tool_calls, sender, messages
             update_payload['awaiting_content'] = True
         if monitor_snapshot_after:
             update_payload['monitor_snapshot_after'] = monitor_snapshot_after
-            cache_monitor_snapshot(tool_display_id, 'after', monitor_snapshot_after)
+            cache_monitor_snapshot(tool_display_id, 'after', monitor_snapshot_after, username=username)
 
         sender('update_action', update_payload)
 

server/context.py

@@ -640,6 +640,13 @@ def apply_conversation_overrides(terminal: WebTerminal, workspace, conversation_
         meta = data.get("metadata") or {}
         prompt_name = meta.get("custom_prompt_name")
         personalization_name = meta.get("personalization_name")
+        # 安全：元数据中的名称必须过资源名校验，防存储型路径穿越
+        import re as _re
+        def _safe_name(v):
+            v = (v or "").strip()
+            return v if _re.fullmatch(r"[A-Za-z0-9_-]{1,64}", v) else None
+        prompt_name = _safe_name(prompt_name)
+        personalization_name = _safe_name(personalization_name)
         # prompt override
         if prompt_name:
             prompt_path = Path(workspace.data_dir) / "prompts" / f"{prompt_name}.txt"

server/conversation.py

@@ -57,6 +57,7 @@ from modules.personalization_manager import (
     save_personalization_config,
 )
 from modules.upload_security import UploadSecurityError
+from server.security import rate_limited
 from modules.user_manager import UserWorkspace
 from modules.usage_tracker import QUOTA_DEFAULTS
 from modules.sub_agent import TERMINAL_STATUSES
@@ -1220,12 +1221,14 @@ def download_conversation_media(media_id, terminal: WebTerminal, workspace: User
             return jsonify({"success": False, "error": tr("conversation.media_file_not_found")}), 404
 
         mime_type = str(entry.get("mime_type") or "application/octet-stream").strip() or "application/octet-stream"
+        # 可执行/脚本类 mime 一律强制下载，防止存储型 XSS（与 file/content 的 SVG 策略一致）
+        _dangerous_inline = mime_type in {"text/html", "image/svg+xml", "application/xhtml+xml"} or mime_type.startswith("text/html")
         blob_name = str(entry.get("blob_rel_path") or "")
         filename = Path(blob_name).name if blob_name else target_id.replace(":", "_")
         return send_file(
             BytesIO(payload),
-            mimetype=mime_type,
-            as_attachment=False,
+            mimetype="application/octet-stream" if _dangerous_inline else mime_type,
+            as_attachment=_dangerous_inline,
             download_name=filename,
             conditional=True,
             etag=True,
@@ -1724,6 +1727,7 @@ def restore_conversation_versioning_checkpoint(conversation_id, terminal: WebTer
 @conversation_bp.route('/api/conversations/<conversation_id>/compress', methods=['POST'])
 @api_login_required
 @with_terminal
+@rate_limited("conversation_compress", 5, 300, scope="user")
 def compress_conversation(conversation_id, terminal: WebTerminal, workspace: UserWorkspace, username: str):
     """深层压缩指定对话（in-place）：生成 compact 文件、标记历史前缀为已压缩，按设置决定是否续接。"""
     try:

server/files.py

@@ -5,8 +5,6 @@
 """
 from __future__ import annotations
 import os
-import zipfile
-from io import BytesIO
 from pathlib import Path
 from typing import Dict, Any, List, Optional, Tuple
 
@@ -78,16 +76,9 @@ def gui_download_entry(terminal, workspace, username):
     try:
         target = manager.prepare_download(path)
         if target.is_dir():
-            memory_file = BytesIO()
-            with zipfile.ZipFile(memory_file, mode='w', compression=zipfile.ZIP_DEFLATED) as zf:
-                for root, dirs, files in os.walk(target):
-                    for file in files:
-                        full_path = Path(root) / file
-                        arcname = manager._to_relative(full_path)
-                        zf.write(full_path, arcname=arcname)
-            memory_file.seek(0)
-            download_name = f"{target.name}.zip"
-            return send_file(memory_file, as_attachment=True, download_name=download_name, mimetype='application/zip')
+            # 文件夹打包下载已下线（2026-09-02 安全审计：os.walk/zipfile 对文件型符号链接
+            # 会读取链接目标内容，容器内可借此泄露宿主任意文件）
+            return jsonify({"success": False, "error": tr("files.folder_download_removed")}), 410
         return send_file(target, as_attachment=True, download_name=target.name)
     except Exception as exc:
         return jsonify({"success": False, "error": str(exc)}), 400

server/monitor.py

@@ -7,12 +7,22 @@ from .state import MONITOR_SNAPSHOT_CACHE, MONITOR_SNAPSHOT_CACHE_LIMIT
 __all__ = ["cache_monitor_snapshot", "get_cached_monitor_snapshot"]
 
 
-def cache_monitor_snapshot(execution_id: Optional[str], stage: str, snapshot: Optional[Dict[str, Any]]):
+def _cache_key(execution_id: Optional[str], username: Optional[str]) -> Optional[str]:
+    """快照缓存键带用户维度，防止跨用户读取他人工具执行快照。"""
+    if not execution_id:
+        return None
+    return f"{username or ''}:{execution_id}"
+
+
+def cache_monitor_snapshot(execution_id: Optional[str], stage: str, snapshot: Optional[Dict[str, Any]], username: Optional[str] = None):
     """缓存工具执行前/后的文件快照。"""
     if not execution_id or not snapshot or not snapshot.get('content'):
         return
+    cache_key = _cache_key(execution_id, username)
+    if not cache_key:
+        return
     normalized_stage = 'after' if stage == 'after' else 'before'
-    entry = MONITOR_SNAPSHOT_CACHE.get(execution_id) or {
+    entry = MONITOR_SNAPSHOT_CACHE.get(cache_key) or {
         'before': None,
         'after': None,
         'path': snapshot.get('path'),
@@ -25,7 +35,7 @@ def cache_monitor_snapshot(execution_id: Optional[str], stage: str, snapshot: Op
     }
     entry['path'] = snapshot.get('path') or entry.get('path')
     entry['timestamp'] = time.time()
-    MONITOR_SNAPSHOT_CACHE[execution_id] = entry
+    MONITOR_SNAPSHOT_CACHE[cache_key] = entry
     if len(MONITOR_SNAPSHOT_CACHE) > MONITOR_SNAPSHOT_CACHE_LIMIT:
         try:
             oldest_key = min(
@@ -37,10 +47,11 @@ def cache_monitor_snapshot(execution_id: Optional[str], stage: str, snapshot: Op
             pass
 
 
-def get_cached_monitor_snapshot(execution_id: Optional[str], stage: str) -> Optional[Dict[str, Any]]:
-    if not execution_id:
+def get_cached_monitor_snapshot(execution_id: Optional[str], stage: str, username: Optional[str] = None) -> Optional[Dict[str, Any]]:
+    cache_key = _cache_key(execution_id, username)
+    if not cache_key:
         return None
-    entry = MONITOR_SNAPSHOT_CACHE.get(execution_id)
+    entry = MONITOR_SNAPSHOT_CACHE.get(cache_key)
     if not entry:
         return None
     normalized_stage = 'after' if stage == 'after' else 'before'

server/multi_agent.py

@@ -24,6 +24,7 @@ from modules.multi_agent.role_store import (
     delete_custom_role,
     is_preset_role,
     sync_preset_roles,
+    validate_role_id,
 )
 from config.paths import CUSTOM_ROLES_DIR, WEB_PRESET_ROLES_DIR
 from config.limits import REASONING_EFFORT_LEVELS
@@ -121,6 +122,10 @@ def create_role_api():
         model_key = str(data.get("model_key") or "").strip() or None
         if not role_id or not name or not body_prompt:
             return jsonify({"success": False, "error": tr("multi_agent_api.role_fields_required")}), 400
+        try:
+            role_id = validate_role_id(role_id)
+        except ValueError as exc:
+            return jsonify({"success": False, "error": str(exc)}), 400
         if thinking_mode not in {"fast", "thinking"}:
             thinking_mode = "fast"
         # 不允许覆盖预设角色
@@ -150,6 +155,10 @@ def create_role_api():
 @api_login_required
 def update_role_api(role_id: str):
     """更新用户自定义角色。不能更新预设角色。"""
+    try:
+        role_id = validate_role_id(role_id)
+    except ValueError as exc:
+        return jsonify({"success": False, "error": str(exc)}), 400
     try:
         data = request.get_json() or {}
         runtime_dir, custom_dir = _get_role_dirs()
@@ -183,6 +192,10 @@ def update_role_api(role_id: str):
 @api_login_required
 def delete_role_api(role_id: str):
     """删除用户自定义角色。不能删除预设角色。"""
+    try:
+        role_id = validate_role_id(role_id)
+    except ValueError as exc:
+        return jsonify({"success": False, "error": str(exc)}), 400
     try:
         if is_preset_role(role_id):
             return jsonify({"success": False, "error": tr("multi_agent_api.cannot_delete_preset_role", role_id=role_id)}), 403

server/security.py

@@ -1,6 +1,7 @@
 """安全相关工具：限流、CSRF、Socket Token、工具结果压缩等。"""
 from __future__ import annotations
 import hmac
+import os
 import secrets
 import time
 from typing import Dict, Any, Optional, Tuple
@@ -9,13 +10,34 @@ from functools import wraps
 
 from . import state
 
+
+def _trusted_proxies() -> frozenset:
+    """可信反向代理地址集合（仅这些来源的 X-Forwarded-For 才被采信）。
+
+    默认不信任任何代理：服务直连时 XFF 可被客户端任意伪造，限流/锁定全部失效。
+    部署在 nginx/CF 等反代之后时，用环境变量 ASTRION_TRUSTED_PROXIES 配置，
+    逗号分隔，如 "127.0.0.1,::1"。
+    """
+    raw = os.environ.get("ASTRION_TRUSTED_PROXIES", "")
+    return frozenset(x.strip() for x in raw.split(",") if x.strip())
+
+
+_TRUSTED_PROXIES = _trusted_proxies()
+
+# 内存限流桶/失败计数表的硬性上限，防止伪造来源导致内存无限增长（反向 DoS）。
+_RATE_LIMIT_MAX_KEYS = 20000
+_FAILURE_TRACKER_MAX_KEYS = 20000
+
+
 # 便捷别名
 def get_client_ip() -> str:
-    """获取客户端IP，支持 X-Forwarded-For."""
-    forwarded = request.headers.get("X-Forwarded-For")
-    if forwarded:
-        return forwarded.split(",")[0].strip()
-    return request.remote_addr or "unknown"
+    """获取客户端IP。仅当直接对端是可信代理时才采信 X-Forwarded-For。"""
+    remote = request.remote_addr or "unknown"
+    if remote in _TRUSTED_PROXIES:
+        forwarded = request.headers.get("X-Forwarded-For")
+        if forwarded:
+            return forwarded.split(",")[0].strip() or remote
+    return remote
 
 
 def resolve_identifier(scope: str = "ip", identifier: Optional[str] = None, kwargs: Optional[Dict[str, Any]] = None) -> str:
@@ -44,9 +66,38 @@ def check_rate_limit(action: str, limit: int, window_seconds: int, identifier: O
         retry_after = window_seconds - int(now - bucket[0])
         return True, max(retry_after, 1)
     bucket.append(now)
+    _gc_rate_limit_buckets(now)
     return False, 0
 
 
+def _gc_rate_limit_buckets(now: float) -> None:
+    """限流桶回收：超过硬上限时清掉已空置/全部过期的桶，防伪造来源刷爆内存。"""
+    buckets = state.RATE_LIMIT_BUCKETS
+    if len(buckets) <= _RATE_LIMIT_MAX_KEYS:
+        return
+    for key in list(buckets.keys()):
+        bucket = buckets.get(key)
+        if not bucket:
+            buckets.pop(key, None)
+    if len(buckets) > _RATE_LIMIT_MAX_KEYS:
+        # 极端情况下（全部桶都活跃）直接整体清空，宁可误伤正常限流状态也不撑爆内存
+        buckets.clear()
+
+
+def gc_failure_trackers() -> None:
+    """失败计数表回收：超过硬上限时清掉已解除锁定的条目。"""
+    trackers = state.FAILURE_TRACKERS
+    if len(trackers) <= _FAILURE_TRACKER_MAX_KEYS:
+        return
+    now = time.time()
+    for key in list(trackers.keys()):
+        entry = trackers.get(key) or {}
+        if not entry.get("blocked_until") or entry["blocked_until"] <= now:
+            trackers.pop(key, None)
+    if len(trackers) > _FAILURE_TRACKER_MAX_KEYS:
+        trackers.clear()
+
+
 def rate_limited(action: str, limit: int, window_seconds: int, scope: str = "ip", error_message: Optional[str] = None):
     """装饰器：为路由增加速率限制。"""
     def decorator(func):
@@ -68,6 +119,7 @@ def rate_limited(action: str, limit: int, window_seconds: int, scope: str = "ip"
 
 def register_failure(action: str, limit: int, lock_seconds: int, scope: str = "ip", identifier: Optional[str] = None, kwargs: Optional[Dict[str, Any]] = None) -> int:
     """记录失败次数，超过阈值后触发锁定。"""
+    gc_failure_trackers()
     ident = resolve_identifier(scope, identifier, kwargs)
     key = f"{action}:{ident}"
     now = time.time()
@@ -204,6 +256,18 @@ def compact_web_search_result(result_data: Dict[str, Any]) -> Dict[str, Any]:
 
 def attach_security_hooks(app):
     """注册 CSRF 校验与通用安全响应头。"""
+    @app.before_request
+    def _block_admin_static_for_non_admin():
+        # Flask 内置 static 路由会优先于蓝图里的 /static/<path> 拦截，
+        # 因此 admin_dashboard 静态资源必须在 before_request 层面拦截。
+        path = request.path or ""
+        if path.startswith("/static/admin_dashboard"):
+            from .auth_helpers import get_current_user_record  # 局部导入避免循环
+            user = get_current_user_record()
+            if not user or getattr(user, "role", None) != "admin":
+                from flask import abort
+                abort(404)
+
     @app.before_request
     def _enforce_csrf_token():
         method = (request.method or "GET").upper()

server/socket_handlers.py

@@ -345,9 +345,36 @@ def handle_message(data):
     start_chat_task(terminal, message, images, send_with_activity, client_sid, workspace, username, videos)
 
 
+# WS 客户端日志事件的轻量限流（防已认证用户洪泛写盘；username -> 时间戳滑窗）
+_WS_CLIENT_LOG_LIMIT = 30  # 每 60 秒最多 30 条
+_WS_CLIENT_LOG_WINDOW = 60.0
+_ws_client_log_buckets: Dict[str, list] = {}
+_WS_CLIENT_LOG_MAX_KEYS = 10000
+
+
+def _ws_client_log_allowed(username: str) -> bool:
+    now = time.time()
+    if len(_ws_client_log_buckets) > _WS_CLIENT_LOG_MAX_KEYS:
+        # 全局回收：清空过老桶（简单策略，防止桶表无限膨胀）
+        for key in [k for k, v in _ws_client_log_buckets.items() if not v or now - v[-1] > _WS_CLIENT_LOG_WINDOW]:
+            _ws_client_log_buckets.pop(key, None)
+        if len(_ws_client_log_buckets) > _WS_CLIENT_LOG_MAX_KEYS:
+            _ws_client_log_buckets.clear()
+    bucket = _ws_client_log_buckets.setdefault(username, [])
+    while bucket and now - bucket[0] > _WS_CLIENT_LOG_WINDOW:
+        bucket.pop(0)
+    if len(bucket) >= _WS_CLIENT_LOG_LIMIT:
+        return False
+    bucket.append(now)
+    return True
+
+
 @socketio.on('client_chunk_log')
 def handle_client_chunk_log(data):
     """前端chunk日志上报"""
+    username = connection_users.get(request.sid)
+    if not username or not _ws_client_log_allowed(username):
+        return
     conversation_id = data.get('conversation_id')
     chunk_index = int(data.get('index') or data.get('chunk_index') or 0)
     elapsed = float(data.get('elapsed') or 0.0)
@@ -359,6 +386,9 @@ def handle_client_chunk_log(data):
 @socketio.on('client_stream_debug_log')
 def handle_client_stream_debug_log(data):
     """前端流式调试日志"""
+    username = connection_users.get(request.sid)
+    if not username or not _ws_client_log_allowed(username):
+        return
     if not isinstance(data, dict):
         return
     entry = dict(data)

server/status/app.py

@@ -74,6 +74,15 @@ def get_app_version_info():
 
 @status_bp.route('/api/app/apk/latest')
 def download_latest_apk():
+    # 安全：未认证全量下载 130MB+ 文件是带宽放大攻击面（2026-09-02 审计）；
+    # 与 /api/app/version 一致要求登录，并加来源限流。
+    from server.auth_helpers import is_logged_in
+    if not is_logged_in():
+        return jsonify({"success": False, "error": tr("auth.not_logged_in")}), 401
+    from server.security import check_rate_limit, get_client_ip
+    limited, retry_after = check_rate_limit("apk_download", 10, 60, get_client_ip())
+    if limited:
+        return jsonify({"success": False, "error": "rate_limited", "retry_after": retry_after}), 429
     project_root = Path(__file__).resolve().parent.parent.parent
     apk_path = _resolve_android_apk_path(project_root)
     if not apk_path.exists():

server/status/base.py

@@ -21,7 +21,7 @@ from server.state import (
     container_manager,
     user_manager,
 )
-from config import AGENT_VERSION, TERMINAL_SANDBOX_MODE
+from config import AGENT_VERSION, TERMINAL_SANDBOX_MODE, TERMINAL_SANDBOX_MOUNT_PATH
 from modules.host_workspace_manager import (
     create_host_workspace,
     delete_host_workspace,
@@ -149,7 +149,12 @@ def get_status(terminal, workspace, username):
     except Exception as exc:
         log_conn_diag(f"status conversation-meta-failed user={username} error={exc}")
     timings["conversation_meta_ms"] = (time.perf_counter() - phase_started_at) * 1000
-    status['project_path'] = str(workspace.project_path)
+    # docker/web 多用户模式下，宿主绝对路径会泄露服务器目录布局与系统用户名，
+    # 对用户无实际意义（操作均在容器 /workspace 内），脱敏为容器视角路径。
+    if _is_host_mode_request():
+        status['project_path'] = str(workspace.project_path)
+    else:
+        status['project_path'] = TERMINAL_SANDBOX_MOUNT_PATH
     phase_started_at = time.perf_counter()
     try:
         # 首屏状态只需要容器是否存在/运行等轻量信息；Docker stats 很慢，

server/status/file_open.py

@@ -327,6 +327,10 @@ def _open_file_with_app(file_path: Path, app_id: str) -> bool:
 @api_login_required
 @with_terminal
 def open_project_in_file_manager(terminal, workspace, username):
+    # 在宿主 GUI 上弹文件管理器窗口——仅 host（单机）模式合理；
+    # docker/web 多用户模式下这会让任何登录用户在服务器桌面弹窗（骚扰 + 路径存在性探测）
+    if not _is_host_mode_request():
+        return jsonify({"success": False, "error": tr("status_file_open.host_mode_only")}), 403
     data = request.get_json(silent=True) or {}
     target_workspace_id = (data.get("workspace_id") or "").strip()
 

server/tasks/api.py

@@ -17,6 +17,7 @@ from flask import current_app, session
 from server.auth_helpers import api_login_required, get_current_username
 from server.context import get_user_resources, ensure_conversation_loaded
 from server.chat_flow import run_chat_task_sync
+from server.security import rate_limited
 from server.state import stop_flags
 from server.utils_common import debug_log, log_conn_diag
 from utils.host_workspace_debug import write_host_workspace_debug
@@ -109,11 +110,15 @@ def get_conversation_running_status_api(conversation_id: str):
 
 @tasks_bp.route("/api/tasks", methods=["POST"])
 @api_login_required
+@rate_limited("chat_task_create", 30, 60, scope="user")
 def create_task_api():
     username = get_current_username()
     workspace_id = session.get("workspace_id") or "default"
     payload = request.get_json() or {}
     message = (payload.get("message") or "").strip()
+    from config import MAX_MESSAGE_CHARS
+    if len(message) > MAX_MESSAGE_CHARS:
+        return jsonify({"success": False, "error": tr("tasks.message_too_long")}), 400
     images, videos = _normalize_media_payload(payload.get("images") or [], payload.get("videos") or [])
     files = _normalize_files_payload(payload.get("files"))
     conversation_id = payload.get("conversation_id")

static/src/admin/apiMain.ts

@@ -1,4 +1,7 @@
 import { createApp } from 'vue';
+import { installI18n } from '@/locales';
 import ApiAdminApp from './ApiAdminApp.vue';
 
-createApp(ApiAdminApp).mount('#admin-api-app');
+const app = createApp(ApiAdminApp);
+installI18n(app);
+app.mount('#admin-api-app');

static/src/admin/customToolsGuideMain.ts

@@ -1,4 +1,7 @@
 import { createApp } from 'vue';
+import { installI18n } from '@/locales';
 import CustomToolsGuideApp from './CustomToolsGuideApp.vue';
 
-createApp(CustomToolsGuideApp).mount('#custom-tools-guide');
+const app = createApp(CustomToolsGuideApp);
+installI18n(app);
+app.mount('#custom-tools-guide');

static/src/admin/customToolsMain.ts

@@ -1,4 +1,7 @@
 import { createApp } from 'vue';
+import { installI18n } from '@/locales';
 import CustomToolsApp from './CustomToolsApp.vue';
 
-createApp(CustomToolsApp).mount('#custom-tools-app');
+const app = createApp(CustomToolsApp);
+installI18n(app);
+app.mount('#custom-tools-app');

static/src/admin/main.ts

@@ -1,4 +1,7 @@
 import { createApp } from 'vue';
+import { installI18n } from '@/locales';
 import AdminDashboardApp from './AdminDashboardApp.vue';
 
-createApp(AdminDashboardApp).mount('#admin-app');
+const app = createApp(AdminDashboardApp);
+installI18n(app);
+app.mount('#admin-app');

static/src/admin/policyMain.ts

@@ -1,4 +1,7 @@
 import { createApp } from 'vue';
+import { installI18n } from '@/locales';
 import PolicyApp from './PolicyApp.vue';
 
-createApp(PolicyApp).mount('#admin-policy-app');
+const app = createApp(PolicyApp);
+installI18n(app);
+app.mount('#admin-policy-app');

utils/conversation_manager/metadata_mixin.py

@@ -50,7 +50,10 @@ class MetadataMixin:
         return f"conv_{timestamp}_{ms:03d}"
 
     def _get_conversation_file_path(self, conversation_id: str) -> Path:
-        """获取对话文件路径"""
+        """获取对话文件路径（白名单校验 ID，防止路径穿越读写数据目录外的文件）。"""
+        import re
+        if not re.fullmatch(r"conv_[A-Za-z0-9_-]+", conversation_id or ""):
+            raise ValueError(tr("conversation.invalid_conversation_id"))
         return self.conversations_dir / f"{conversation_id}.json"
 
     def _extract_title_from_messages(self, messages: List[Dict]) -> str: