Data Formulator 服务端路径安全开发规范:ConfinedDir 路径约束原语与全链路防护实践
【免费下载链接】data-formulator🪄 Data Formulator is an interactive AI-powered data analysis system makes it easy to connect, explore and visualize data.项目地址: https://gitcode.com/GitHub_Trending/da/data-formulator
本文基于
docs/dev-guides/8-path-safety.md(Data Formulator 核心团队维护,最后更新 2026-04-28)整理并扩展,结合仓库源码与测试用例深入讲解服务端路径安全的设计与落地。
导读
Data Formulator 是交互式 AI 数据分析系统,后端同时面向用户上传、LLM 生成的 Agent 工具参数、外部存储与 HTTP 请求等多元输入。任何把"不可信路径片段"拼到"服务端根目录"上的代码,都必须先经过统一路径约束。本文是面向后端开发者的路径安全规范,系统讲解ConfinedDir路径约束原语的设计与四层防护、文件名清洗的两层防线、文件下载/上传 Route、Agent 工具、Data Loader 与 Sandbox 部署的安全要求,以及对应的测试与迁移清单。读完本文,你将掌握 Data Formulator 中统一的路径安全接入方式,能够正确使用ConfinedDir避免手写resolve() + relative_to()这类易出错的反模式,并懂得如何为新增模块做路径安全检查。
1. 核心原则:所有路径操作必须经过统一约束
Data Formulator 的路径安全体系建立在一条铁律之上:任何把"不可信路径片段"拼到"服务端根目录"上的代码,都必须先经过统一路径约束。这里的"不可信路径片段"来源包括:
- 用户输入(上传文件名、URL、表单参数);
- LLM 生成的 Agent 工具参数(LLM 输入又来自用户,因此同样不可信);
- 外部存储返回的对象名(如 Blob 路径);
- HTTP body / query / path 参数。
下表汇总了各类场景对应的标准做法(即"应当怎么做"):
| 场景 | 规范 |
|---|---|
| Workspace data 文件 | Workspace.get_file_path()(内部使用ConfinedDir) |
| Workspace 根/data/scratch 子目录 | workspace.confined_root/confined_data/confined_scratch属性 |
| Agent 读文件/列目录工具 | workspace.confined_root/workspace.confined_scratch,传入各工具方法 |
| 文件下载 route | workspace.confined_scratch.resolve(filename)后传给send_file() |
| 文件上传 route | secure_filename()清洗 +workspace.confined_scratch.resolve()二次校验 |
| 任意 root + relative path | ConfinedDir(root).resolve(relative) |
| 知识库文件读写 | 通过KnowledgeStore内部的ConfinedDir |
| 推理日志写入 | 通过ReasoningLogger内部的ConfinedDir |
| 读取宿主文件系统的 Loader | 必须注册多用户部署禁用规则 |
| Sandbox 部署 | 多用户模式不得使用not_a_sandbox |
同时明确两条绝对禁止:
- 禁止手写
resolve() + relative_to()或resolve() + is_relative_to()路径检查模式。这些逻辑已统一封装在ConfinedDir.resolve()中,手写会导致逻辑重复、不一致和遗漏风险; - 禁止把用户、LLM、外部存储、HTTP 参数直接用于裸路径拼接,例如
Path(root) / user_input、root / filename。
local_folder_data_loader.py是全仓库首个采用ConfinedDir的实现(即"原始采用者"),后续的迁移都以此为标准逐步推进。
2. 路径约束原语:ConfinedDir 的四层防护
ConfinedDir是服务端路径约束的默认原语,位于 path_safety.py。它实现了一个"目录监狱"(directory jail):所有路径解析都通过这一个唯一关口(single chokepoint)进行,一旦解析结果逃出 root,立即抛出ValueError。
2.1 基本用法
from data_formulator.security.path_safety import ConfinedDir jail = ConfinedDir(tmp_path, mkdir=False) target = jail.resolve("data/report.csv") jail.write("scratch/output.csv", b"content")从源码看(path_safety.py),构造时ConfinedDir会先对 root 做Path(root).resolve()归一化,mkdir=True(默认)时还会自动创建目录树。实例在构造后不可变(使用__slots__仅存_root),因此是线程安全的——文档注释明确说明Path.resolve()与is_relative_to()属于 OS 层操作,天然适合并发使用。
2.2 resolve() 的四层防护
ConfinedDir.resolve()的防护层次(见 path_safety.py):
- 拒绝空路径:
relative为空直接抛ValueError("Empty relative path"); - 拒绝绝对路径:
rel.is_absolute() or rel.root为真即拒绝,防止Path(root) / "/etc/passwd"之类拼接意外覆盖 root; - 拒绝
..路径段:显式检查Path(relative).parts中是否存在"..",在拼接之前就把路径穿越段挡下; - 符号链接逃逸检查:
(self._root / relative).resolve()展开所有符号链接后,用Path.is_relative_to()确认规范化结果仍在 root 内——这一步专门防 symlink escape。
捕获ValueError时,调用方应返回应用层错误或跳过不可信外部对象,不要继续使用原始路径。
2.3 扩展 API
除resolve()与write()外,ConfinedDir还提供了一组扩展方法,供推理日志(ReasoningLogger)、知识库(KnowledgeStore)等场景复用,所有操作都经由resolve()继承路径穿越防护:
| 方法 | 用途 | 关键行为 |
|---|---|---|
read_text(relative, encoding="utf-8") | 读文本文件 | 默认 UTF-8 |
write_text(relative, content, encoding="utf-8") | 写文本文件 | 自动创建父目录 |
exists(relative) | 判断存在性 | 穿越路径返回False而非抛错,调用方可视为"不存在" |
iterdir(relative="") | 列目录 | 空字符串表示 root 本身 |
rglob(pattern, relative="") | 递归 glob | 起始点同样受约束 |
unlink(relative) | 删除文件 | 穿越路径抛ValueError |
__truediv__运算符重载 | jail / "sub/path" | 等价于jail.resolve("sub/path") |
对应行为均有测试覆盖,见 test_confined_dir_extended.py:例如read_text("../etc/passwd")抛ValueError;exists("../../../etc/passwd")返回False;write_text("../../evil.txt", ...)抛ValueError;jail / "x.txt"返回与(tmp_path / "x.txt").resolve()相等的路径。
3. 文件名清洗:两层独立防线
路径约束和文件名清洗是两层不同防线,职责各异、缺一不可:
| API | 用途 |
|---|---|
safe_data_filename() | Workspace 数据文件名,保留 Unicode(中文、日文、韩文等),去掉目录组件和控制字符 |
secure_filename() | identity、URL/上传临时文件名等需要 ASCII 安全名的场景 |
ConfinedDir.resolve() | 校验清洗后的相对路径不会逃出 root |
safe_data_filename()实现在 parquet_utils.py:通过Path(filename).name提取 basename 从而去掉所有目录组件,再用正则[\x00-\x1f]剥离控制字符,最后拒绝空名、.与..。它之所以保留 Unicode,是因为werkzeug.secure_filename会直接剥离非 ASCII 字符,不适合中文/日文文件名场景;而secure_filename()则用于 identity 目录名(见下文)、URL、上传临时文件等需要纯 ASCII 安全名的场景。
关键实践:使用Workspace.get_file_path(filename)的场景不需要手动调用ConfinedDir。因为该方法内部已经实现了两层防御(见 workspace.py):
def get_file_path(self, filename: str) -> Path: basename = safe_data_filename(filename) # 第一层:清洗 try: return self._confined_data.resolve(basename) # 第二层:约束 except ValueError: raise ValueError(f"Path traversal detected: {filename!r}")回归测试 test_confined_dir_migration.py 的test_get_file_path_traversal_sanitized验证了这一协作机制:get_file_path("../../etc/passwd")经safe_data_filename清洗为"passwd"(第一层),再经ConfinedDir校验(第二层),最终结果安全落在data/子目录内。
identity 目录名同样需要清洗:Workspace.__init__通过sanitize_identity_dirname()(workspace.py)用secure_filename产出安全的单段目录名,长度超过 256 或清洗结果为空时抛ValueError;随后再用ConfinedDir(self._root, mkdir=False).resolve(self._safe_id)二次确认构造出的路径没有逃出根目录(workspace.py)。
4. 文件下载 Route:安全检查与实际发送必须共用同一路径
下载接口必须让安全检查和实际发送使用同一个 resolved path。Data Formulator 的 scratch 下载路由scratch_serve(routes/agents.py)示范了标准写法:
from flask import send_file scratch_jail = workspace.confined_scratch try: target = scratch_jail.resolve(filename) except ValueError: return jsonify(status="error", message="Access denied") return send_file(target)源码中实际实现是通过AppError(ErrorCode.ACCESS_DENIED, "Access denied")返回统一错误(对应前端错误码体系),随后检查target.exists(),不存在时抛TABLE_NOT_FOUND。
❌ 禁止手写 resolve + relative_to 检查(已弃用模式):
# BAD — 手写检查,已弃用 target = (scratch_dir / filename).resolve() if not target.is_relative_to(scratch_dir.resolve()): return jsonify(status="error", message="Access denied")禁止在用户路径上使用send_from_directory(dir, filename)。它会在 Flask 内部再次解析原始filename,容易和前置安全检查形成 TOCTOU(检查时间与使用时间不一致)漏洞——两次解析之间文件名指向可能已经改变。
如果文件名来自用户输入并写入Content-Disposition,不得直接插值原始字符串。新代码应先建立统一 helper:去除 CR/LF、引号和目录组件,并为非 ASCII 名称提供安全 fallback 或filename*(RFC 5987)编码。
上传路由的"清洗 + 二次校验"组合
scratch 上传路由scratch_upload(routes/agents.py)体现了双层防线:先用werkzeug.secure_filename清洗原始文件名,再拼接内容 hash 生成{base}_{file_hash}{ext}形式的最终文件名,最后用scratch_jail.resolve(final_name)二次校验。测试test_scratch_upload_traversal_sanitized(test_confined_dir_migration.py)验证:上传名为"../../etc/passwd"的文件,最终写入路径仍位于 scratch 目录内。
5. Agent 工具:LLM 生成的路径参数一律视为不可信
Agent 工具参数由 LLM 生成,LLM 输入又来自用户,因此路径参数一律视为不可信。入口函数应使用Workspace.confined_*属性获取ConfinedDir实例,传入各工具方法(见 agent_data_loading_chat.py):
def _execute_tool(self, name, args): workspace_jail = self.workspace.confined_root scratch_jail = self.workspace.confined_scratch if name == "read_file": return self._tool_read_file(args, workspace_jail) elif name == "write_file": return self._tool_write_file(args, scratch_jail) elif name == "list_directory": return self._tool_list_directory(args, workspace_jail) elif name == "execute_python": return self._tool_execute_python(args) elif name == "fetch_url": return self._tool_fetch_url(args, scratch_jail) ...具体工具通过ConfinedDir.resolve()获取安全路径:
def _tool_read_file(self, args, workspace_jail): rel_path = args.get("path", "") try: target = workspace_jail.resolve(rel_path) except ValueError: return {"error": "Access denied: path outside workspace"} ...从源码看各工具的 jail 分配逻辑(agent_data_loading_chat.py):
_tool_read_file/_tool_list_directory:使用workspace_jail(confined_root),可读 workspace 内任意文件;_tool_write_file:使用scratch_jail(confined_scratch),并先经_secure_filename清洗;_tool_execute_python:DataFrame 自动保存到scratch_jail.resolve(f"{safe_name}.csv");_preview_scratch_files:使用workspace.confined_root.resolve(file_path)读取 scratch CSV 构建预览 action。
❌ 禁止在工具方法中手写resolve() + relative_to()(已弃用模式):
# BAD — 手写检查,已弃用 target = (workspace_path / rel_path).resolve() try: target.relative_to(workspace_path) except ValueError: return {"error": "Access denied: path outside workspace"}性能与一致性约定:不要在工具函数内部反复创建ConfinedDir或反复调用resolve()。在_execute_tool入口创建一次,所有工具方法复用同一个实例。对应测试见 test_confined_dir_migration.py(TestAgentToolsUseConfinedProperties):_tool_read_file对"../../etc/passwd"返回含 "Access denied" 的 error;_tool_write_file对"../../evil.txt"的写入被重定向到 scratch 目录内。
6. Data Loader 与宿主文件系统:多用户模式必须禁用
如果 Loader 的构造参数包含用户可控的本机路径,例如root_dir,它只能在本地单用户模式使用。原因很直接:这类 Loader 直接读取服务器宿主文件系统,多用户场景下会成为任意文件读取的后门。
必须在 data_loader/init.py 的_enforce_deployment_restrictions()中注册禁用规则:
def _enforce_deployment_restrictions(): backend = os.environ.get("WORKSPACE_BACKEND", "local") if backend != "local" and "your_local_loader" in DATA_LOADERS: del DATA_LOADERS["your_local_loader"] DISABLED_LOADERS["your_local_loader"] = ( "your_local_loader connector is disabled in multi-user mode " "(WORKSPACE_BACKEND != 'local')" )仓库中的参考实现正是如此:local_folderLoader(local_folder_data_loader.py,即规范中提到的"当前参考实现"与ConfinedDir原始采用者)在WORKSPACE_BACKEND != "local"时从DATA_LOADERS删除并写入DISABLED_LOADERS,附上人类可读的禁用原因(data_loader/init.py)。该模块顶层同时维护DATA_LOADERS(成功导入的 Loader)与DISABLED_LOADERS(失败或禁用项及提示),前端可据此向用户解释缺失原因。
相关背景:该模块的插件扫描机制同样受此约束——
DF_ALLOW_PLUGINS=1显式 opt-in 之前,只有WORKSPACE_BACKEND=local的本地单用户模式才允许扫描执行插件目录(插件是任意 Python 代码,风险更高,见 data_loader/init.py)。
Data Loader 通用开发规范见 3-data-loader-development.md。
7. Sandbox 部署:多用户模式禁止 not_a_sandbox
多用户或云部署中,not_a_sandbox会让 LLM 生成的 Python 代码在宿主进程直接执行,可能绕过所有路径检查。Data Formulator 的沙箱目录(sandbox/)包含三种实现:docker_sandbox(最大隔离)、local_sandbox(隔离子进程 + audit hooks)与not_a_sandbox。
部署要求:
WORKSPACE_BACKEND == "local"时允许桌面单用户模式使用not_a_sandbox;WORKSPACE_BACKEND != "local"时必须使用SANDBOX=docker或SANDBOX=local;- 新部署模板应默认选择隔离沙箱。
应用启动时已有安全检查:app.py中的_safety_checks()(app.py)在检测到multi_user且sandbox == 'not_a_sandbox'时输出 critical 级别告警日志,提示"LLM-generated code can read/write arbitrary files on the server. Set SANDBOX=docker or SANDBOX=local for production deployments."。
注意:当前实现是告警而非硬阻断(不会阻止启动),因此生产部署还必须在部署配置或启动脚本层强制SANDBOX=local/SANDBOX=docker。CLI 启动参数层面,--sandbox的合法取值就是['local', 'docker'](app.py),默认值取自环境变量SANDBOX(默认local),也就是说通过 CLI 参数本身就无法显式选择not_a_sandbox——它只在内部默认分支下出现。
8. 测试要求:路径安全必须有回归测试兜底
新增路径相关代码时,至少覆盖以下用例:
- 正常相对路径可以访问;
../、绝对路径、空路径被拒绝;- symlink escape 被拒绝,适用时用真实文件系统测试;
- 文件下载 route 使用
send_file(resolved_path); - 多用户部署下宿主文件系统 Loader 被禁用。
仓库已有的参考测试(均可直接阅读与复用):
| 测试文件 | 覆盖内容 |
|---|---|
| test_local_folder_loader.py | 宿主文件系统 Loader 的路径安全 |
| test_scratch_serve.py | scratch 下载/上传路由安全 |
| test_tool_path_safety.py | Agent 工具路径安全 |
| test_local_folder_deployment.py | 多用户部署禁用规则 |
| test_startup_safety.py | 启动期安全检查(含 not_a_sandbox 告警) |
| test_confined_dir_extended.py | ConfinedDir扩展 API(read_text/write_text/exists/iterdir/rglob/unlink) |
| test_confined_dir_migration.py | 迁移回归:Workspace.confined_*属性、Agent 工具、scratch 路由 |
值得注意的测试细节:TestScratchRoutesConfinedMigration(test_confined_dir_migration.py)通过 Flask test client 真实请求/api/agent/workspace/scratch/../../../etc/passwd,断言返回 403 与ACCESS_DENIED错误码,属于端到端级别的路径穿越验证;TestWorkspaceConfinedProperties则逐项断言confined_root/confined_data/confined_scratch分别指向 workspace 根、data/、scratch/子目录,并对各自目录发起穿越探测。
9. New Module Checklist:新增模块自检清单
任何新模块在合入前,请逐项核对以下清单:
- 新模块是否接收用户、LLM、外部存储或 HTTP 传入的路径片段?
- 是否使用了
ConfinedDir或Workspace.get_file_path()? - 是否避免了
Path(root) / user_input裸拼接? - 是否避免了手写
resolve() + relative_to()/is_relative_to()模式?(必须用ConfinedDir) - 下载 route 是否用
ConfinedDir.resolve()做检查并传给send_file()? - 上传 route 是否同时使用
secure_filename()和ConfinedDir.resolve()? - 读取宿主文件系统的 Loader 是否注册了多用户禁用规则?
- 多用户部署是否启用了
docker或localsandbox?
10. 已迁移清单:手写检查的历史清理
规范发布时,以下位置已从手写resolve() + relative_to()迁移到ConfinedDir,可作为新代码的参考范例:
| 文件 | 方法 | 迁移方式 |
|---|---|---|
| datalake/workspace.py | __init__ | 创建_confined_root/_confined_data/_confined_scratch,暴露confined_root/confined_data/confined_scratch属性(见 workspace.py) |
datalake/workspace.py | get_file_path | self._confined_data.resolve(basename) |
datalake/workspace.py | __init__(legacy root) | ConfinedDir(root).resolve(safe_id) |
| agent_data_loading_chat.py | _execute_tool | workspace.confined_root+workspace.confined_scratch |
agent_data_loading_chat.py | _tool_read_file | workspace_jail.resolve(rel_path) |
agent_data_loading_chat.py | _tool_list_directory | workspace_jail.resolve(rel_path) |
agent_data_loading_chat.py | _tool_write_file | scratch_jail.resolve(filename) |
agent_data_loading_chat.py | _tool_execute_python | workspace.confined_scratch.resolve(safe_name + ".csv") |
agent_data_loading_chat.py | _preview_scratch_files | workspace.confined_root.resolve(file_path) |
| routes/agents.py | scratch_serve | workspace.confined_scratch.resolve(filename) |
routes/agents.py | scratch_upload | workspace.confined_scratch.resolve(final_name) |
cached_azure_blob_workspace.py | _cache_path | self._cache_jail.resolve(filename) |
| knowledge/store.py | CRUD | ConfinedDir(user_home / "knowledge" / category) |
| agents/reasoning_log.py | log | ConfinedDir(DATA_FORMULATOR_HOME / "agent-logs" / date / safe_identity_id) |
local_folder_data_loader.py | 全文件 | 已使用ConfinedDir(原始采用者) |
补充两处迁移后的实际形态,帮助理解"复用而非重复创建"的原则:
- 知识库:
KnowledgeStore的所有文件 I/O 都经由ConfinedDir(knowledge/store.py),且对目录深度有额外约束——rules类别只允许平铺文件(1 层路径),workflows允许一层子目录(最多 2 层路径),将用户可控的目录结构限制在极小的攻击面上; - 推理日志:
ReasoningLogger按DATA_FORMULATOR_HOME/agent-logs/<date>/<safe_identity_id>/<session_id>-<agent_type>.jsonl组织日志(reasoning_log.py),同样通过ConfinedDir(...).resolve(self._filename, mkdir_parents=True)落盘。
总结
Data Formulator 的路径安全体系可以概括为一句话:一个原语(ConfinedDir)、两层防线(文件名清洗 + 路径约束)、三类场景(Route / Agent 工具 / 存储与日志)、一个自检清单。ConfinedDir将空路径、绝对路径、..段、symlink 逃逸四类攻击统一拦截在resolve()这一唯一关口;safe_data_filename与secure_filename分别面向 Unicode 数据文件名与 ASCII 安全名场景完成前置清洗;Workspace 通过confined_root/confined_data/confined_scratch三个属性向 Agent 与路由暴露受控的访问入口。对于新的后端代码,遵循"入口创建一次 jail、统一 resolve、不做裸拼接、不手写 relative_to"即可平稳接入这套体系,并用仓库中现成的测试文件验证正确性。
【免费下载链接】data-formulator🪄 Data Formulator is an interactive AI-powered data analysis system makes it easy to connect, explore and visualize data.项目地址: https://gitcode.com/GitHub_Trending/da/data-formulator
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考