Data Formulator 服务端路径安全开发规范:ConfinedDir 路径约束原语与全链路防护实践
2026/9/14 0:01:11 网站建设 项目流程

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,传入各工具方法
文件下载 routeworkspace.confined_scratch.resolve(filename)后传给send_file()
文件上传 routesecure_filename()清洗 +workspace.confined_scratch.resolve()二次校验
任意 root + relative pathConfinedDir(root).resolve(relative)
知识库文件读写通过KnowledgeStore内部的ConfinedDir
推理日志写入通过ReasoningLogger内部的ConfinedDir
读取宿主文件系统的 Loader必须注册多用户部署禁用规则
Sandbox 部署多用户模式不得使用not_a_sandbox

同时明确两条绝对禁止

  1. 禁止手写resolve() + relative_to()resolve() + is_relative_to()路径检查模式。这些逻辑已统一封装在ConfinedDir.resolve()中,手写会导致逻辑重复、不一致和遗漏风险;
  2. 禁止把用户、LLM、外部存储、HTTP 参数直接用于裸路径拼接,例如Path(root) / user_inputroot / 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):

  1. 拒绝空路径relative为空直接抛ValueError("Empty relative path")
  2. 拒绝绝对路径rel.is_absolute() or rel.root为真即拒绝,防止Path(root) / "/etc/passwd"之类拼接意外覆盖 root;
  3. 拒绝..路径段:显式检查Path(relative).parts中是否存在"..",在拼接之前就把路径穿越段挡下;
  4. 符号链接逃逸检查(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")ValueErrorexists("../../../etc/passwd")返回Falsewrite_text("../../evil.txt", ...)ValueErrorjail / "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_jailconfined_root),可读 workspace 内任意文件;
  • _tool_write_file:使用scratch_jailconfined_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=dockerSANDBOX=local
  • 新部署模板应默认选择隔离沙箱。

应用启动时已有安全检查:app.py中的_safety_checks()(app.py)在检测到multi_usersandbox == '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.pyscratch 下载/上传路由安全
test_tool_path_safety.pyAgent 工具路径安全
test_local_folder_deployment.py多用户部署禁用规则
test_startup_safety.py启动期安全检查(含 not_a_sandbox 告警)
test_confined_dir_extended.pyConfinedDir扩展 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 传入的路径片段?
  • 是否使用了ConfinedDirWorkspace.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 是否注册了多用户禁用规则?
  • 多用户部署是否启用了dockerlocalsandbox?

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.pyget_file_pathself._confined_data.resolve(basename)
datalake/workspace.py__init__(legacy root)ConfinedDir(root).resolve(safe_id)
agent_data_loading_chat.py_execute_toolworkspace.confined_root+workspace.confined_scratch
agent_data_loading_chat.py_tool_read_fileworkspace_jail.resolve(rel_path)
agent_data_loading_chat.py_tool_list_directoryworkspace_jail.resolve(rel_path)
agent_data_loading_chat.py_tool_write_filescratch_jail.resolve(filename)
agent_data_loading_chat.py_tool_execute_pythonworkspace.confined_scratch.resolve(safe_name + ".csv")
agent_data_loading_chat.py_preview_scratch_filesworkspace.confined_root.resolve(file_path)
routes/agents.pyscratch_serveworkspace.confined_scratch.resolve(filename)
routes/agents.pyscratch_uploadworkspace.confined_scratch.resolve(final_name)
cached_azure_blob_workspace.py_cache_pathself._cache_jail.resolve(filename)
knowledge/store.pyCRUDConfinedDir(user_home / "knowledge" / category)
agents/reasoning_log.pylogConfinedDir(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 层路径),将用户可控的目录结构限制在极小的攻击面上;
  • 推理日志ReasoningLoggerDATA_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_filenamesecure_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),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询