From 298f13721c2cabef06a3f64f3653cd4930cd7ac4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=88=98=E8=88=AA=E5=AE=87?= <3364451258@qq.com> Date: Fri, 10 Jul 2026 15:19:17 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E7=BB=9F=E4=B8=80=20CLI/API=20=E8=B7=AF?= =?UTF-8?q?=E5=BE=84=E5=AE=89=E5=85=A8=E6=A3=80=E6=9F=A5=E4=B8=BA=20is=5Fp?= =?UTF-8?q?ath=5Fwithin=5Fworkspace?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/cli/main.py | 8 ++++---- src/core/security.py | 20 ++++++++++++++++++++ src/server/app.py | 12 ++---------- 3 files changed, 26 insertions(+), 14 deletions(-) diff --git a/src/cli/main.py b/src/cli/main.py index 7ce454b..a1a823c 100644 --- a/src/cli/main.py +++ b/src/cli/main.py @@ -17,7 +17,7 @@ if sys.stdout.encoding != "utf-8": sys.path.insert(0, str(Path(__file__).parent.parent)) from src.core.config import DEFAULT_CONFIG_PATH -from src.core.security import is_safe_path +from src.core.security import is_path_within_workspace from src.server.deps import get_state, get_default_collection app = typer.Typer( @@ -82,7 +82,7 @@ def ingest( if file_paths: total = 0 for fp in file_paths: - if not is_safe_path(fp): + if not is_path_within_workspace(fp): typer.echo(f"[SKIP] 不安全的路径: {fp}", err=True) continue # 支持通配符 (shell 展开或 Python glob) @@ -90,7 +90,7 @@ def ingest( if "*" in fp or "?" in fp: matches = _glob.glob(fp, recursive=True) for m in matches: - if not is_safe_path(m): + if not is_path_within_workspace(m): typer.echo(f"[SKIP] 不安全的路径: {m}", err=True) continue c = ingestor.ingest_file(m) @@ -117,7 +117,7 @@ def ingest_dir( collection: CollectionOpt = None, ): _init_config(config) - if not is_safe_path(dir_path): + if not is_path_within_workspace(dir_path): typer.echo(f"错误: 不安全的路径 — {dir_path}", err=True) raise typer.Exit(code=1) state = get_state() diff --git a/src/core/security.py b/src/core/security.py index 4a856ff..22aaec0 100644 --- a/src/core/security.py +++ b/src/core/security.py @@ -26,3 +26,23 @@ def is_safe_path(path_str: str) -> bool: return False return True + + +def is_path_within_workspace(path_str: str) -> bool: + """检查路径是否在当前工作目录内（防路径穿越 + 目录绑定）. + + 同时检查： + 1. 路径不含 .. 穿越组件且非绝对路径 + 2. resolve 后的路径位于当前工作目录内 + """ + if not is_safe_path(path_str): + return False + + from pathlib import Path + path = Path(path_str).resolve() + cwd = Path.cwd().resolve() + try: + common = Path(os.path.commonpath([str(path), str(cwd)])) + except ValueError: + return False + return common == cwd diff --git a/src/server/app.py b/src/server/app.py index 165f446..4fb45c9 100644 --- a/src/server/app.py +++ b/src/server/app.py @@ -10,7 +10,7 @@ from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import RedirectResponse from pydantic import BaseModel, Field, model_validator -from src.core.security import is_safe_path +from src.core.security import is_path_within_workspace from src.server.auth import verify_api_key, rate_limiter from src.server.deps import get_state, AppState @@ -117,17 +117,9 @@ def ingest_document( ingestor = state.get_ingestor(req.collection) try: if req.file_path: - if not is_safe_path(req.file_path): + if not is_path_within_workspace(req.file_path): raise HTTPException(status_code=400, detail="不允许的路径") path = Path(req.file_path).resolve() - cwd = Path.cwd().resolve() - # 用 commonpath 替代字符串 startswith 比较 (Windows 大小写安全) - try: - common = Path(os.path.commonpath([str(path), str(cwd)])) - except ValueError: - raise HTTPException(status_code=400, detail="不允许访问当前目录外的路径") - if common != cwd: - raise HTTPException(status_code=400, detail="不允许访问当前目录外的路径") if not path.exists(): raise HTTPException(status_code=404, detail=f"文件不存在: {path.name}") count = ingestor.ingest_file(str(path))