fix(security): 增强文件路径验证,防止路径遍历攻击

- 在知识库路由中添加文件路径安全检查,确保上传和下载的文件路径在允许的目录内
- 新增 validate_file_path 函数,规范化并验证文件路径,抛出相应的异常信息
- 更新文档,修复部分已知BUG,提升系统稳定性
This commit is contained in:
Wenjie Zhang 2025-09-23 04:42:04 +08:00
parent a4c33265a2
commit 59c60fa879
3 changed files with 92 additions and 14 deletions

View File

@ -1,5 +1,17 @@
目前已有的开发计划包括:
📝 0.2.2 version
不再新增任何特性,仅作功能调整以及 bug 修复。
🐛**BUGs**
- [ ] 部分 doc 格式的文件支持有问题
- [ ] 当出现不支持的文件类型的时候,前端没有限制
- [ ] 当消息生成的时候有报错的时候,前端无显示(需要解决存储的问题)
---
💭 **Features Todo**
- [x] LangGraph 升级到 0.6+ 版本,并适配新特性,如 context 等,添加 MCP 工具的支持。
- [x] 支持动态工具配置的同时,将 Configuration替换为 Context 后能够正常使用
@ -14,6 +26,7 @@
- [ ] 添加统计信息
- [x] 添加内容审查功能
- [x] 补充 embedding 模型和 reranker 模型的配置说明
- [ ] 支持 MinerU 的解析方法
📝 **Base**
@ -21,18 +34,6 @@
- [ ] 新建 tasker 模块用来管理所有的后台任务UI 上使用侧边栏管理。
- [ ] 新增 files 模块,用来管理文件上传,下载等
🐛**BUGs**
- [x] LlightRAG 知识库中,点击边,没有显示,但是在全屏的时候却又能够显示出来。
- [ ] 部分 doc 格式的文件支持有问题
- [ ] 当出现不支持的文件类型的时候,前端没有限制
- [x] 默认智能体设置后,在一些情况下依然仅加载第一个智能体
- [x] 目前只能获取默认的 tools单个智能体的tools没法展示
- [x] 生成的过程中切换对话,会出现消息渲染混乱的问题,消息完全记载完成之后不会出现混乱
- [ ] 当消息生成的时候有报错的时候,前端无显示(需要解决存储的问题)
- [ ] 偶然会出现 embedding 报错的情况,但不好复现
- [x] GraphCanvas 在智能体消息页面有时候会出现渲染不出现的问题可以添加一个slot 在右上角(加载 button这样就可以强制重新渲染
💯 **More**:
下面的功能**可能**会放在后续版本实现,暂时未定

View File

@ -150,6 +150,15 @@ async def add_documents(
content_type = params.get("content_type", "file")
# 安全检查:验证文件路径
if content_type == "file":
from src.knowledge.kb_utils import validate_file_path
for item in items:
try:
validate_file_path(item, db_id)
except ValueError as e:
raise HTTPException(status_code=403, detail=str(e))
try:
processed_items = await knowledge_base.add_content(db_id, items, params=params)
item_type = "URLs" if content_type == "url" else "files"
@ -222,7 +231,17 @@ async def download_document(db_id: str, doc_id: str, request: Request, current_u
raise HTTPException(status_code=404, detail="File not found")
file_path = file_info.get("meta", {}).get("path")
if not file_path or not os.path.exists(file_path):
if not file_path:
raise HTTPException(status_code=404, detail="File path not found in metadata")
# 安全检查:验证文件路径
from src.knowledge.kb_utils import validate_file_path
try:
normalized_path = validate_file_path(file_path, db_id)
except ValueError as e:
raise HTTPException(status_code=403, detail=str(e))
if not os.path.exists(normalized_path):
raise HTTPException(status_code=404, detail=f"File not found on disk: {file_info=}")
# 获取文件扩展名和MIME类型解码URL编码的文件名
@ -277,7 +296,7 @@ async def download_document(db_id: str, doc_id: str, request: Request, current_u
media_type = media_types.get(ext.lower(), "application/octet-stream")
# 创建自定义FileResponse避免文件名编码问题
response = StarletteFileResponse(path=file_path, media_type=media_type)
response = StarletteFileResponse(path=normalized_path, media_type=media_type)
# 正确处理中文文件名的HTTP头部设置
# HTTP头部只能包含ASCII字符所以需要对中文文件名进行编码

View File

@ -8,6 +8,64 @@ from src import config
from src.utils import hashstr, logger
def validate_file_path(file_path: str, db_id: str = None) -> str:
"""
验证文件路径安全性防止路径遍历攻击
Args:
file_path: 要验证的文件路径
db_id: 数据库ID用于获取知识库特定的上传目录
Returns:
str: 规范化后的安全路径
Raises:
ValueError: 如果路径不安全
"""
try:
# 规范化路径
normalized_path = os.path.abspath(os.path.realpath(file_path))
# 获取允许的根目录
from src.knowledge import knowledge_base
allowed_dirs = [
os.path.abspath(os.path.realpath(config.save_dir)),
]
# 如果指定了db_id添加知识库特定的上传目录
if db_id:
try:
allowed_dirs.append(
os.path.abspath(os.path.realpath(knowledge_base.get_db_upload_path(db_id)))
)
except Exception:
# 如果无法获取db路径使用通用上传目录
allowed_dirs.append(
os.path.abspath(os.path.realpath(os.path.join(config.save_dir, "database", "uploads")))
)
# 检查路径是否在允许的目录内
is_safe = False
for allowed_dir in allowed_dirs:
try:
if normalized_path.startswith(allowed_dir):
is_safe = True
break
except Exception:
continue
if not is_safe:
logger.warning(f"Path traversal attempt detected: {file_path} (normalized: {normalized_path})")
raise ValueError(f"Access denied: Invalid file path: {file_path}")
return normalized_path
except Exception as e:
logger.error(f"Path validation failed for {file_path}: {e}")
raise ValueError(f"Invalid file path: {file_path}")
def split_text_into_chunks(text: str, file_id: str, filename: str, params: dict = {}) -> list[dict]:
"""
将文本分割成块使用 LangChain MarkdownTextSplitter 进行智能分割