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))