fix: add path traversal protection to ingest endpoint
This commit is contained in:
@@ -12,6 +12,16 @@ from src.core.ingest import DocumentIngestor
|
|||||||
from src.core.search import Searcher
|
from src.core.search import Searcher
|
||||||
|
|
||||||
|
|
||||||
|
def _is_safe_path(path_str: str) -> bool:
|
||||||
|
"""检查路径是否安全: 仅允许相对路径且不含 .. 穿越."""
|
||||||
|
normalized = os.path.normpath(path_str)
|
||||||
|
if os.path.isabs(normalized):
|
||||||
|
return False
|
||||||
|
if ".." in normalized.split(os.sep):
|
||||||
|
return False
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
# -- 请求模型 --
|
# -- 请求模型 --
|
||||||
class IngestRequest(BaseModel):
|
class IngestRequest(BaseModel):
|
||||||
file_path: str | None = None
|
file_path: str | None = None
|
||||||
@@ -109,6 +119,8 @@ def ingest_document(req: IngestRequest):
|
|||||||
ingestor = _get_ingestor()
|
ingestor = _get_ingestor()
|
||||||
try:
|
try:
|
||||||
if req.file_path:
|
if req.file_path:
|
||||||
|
if not _is_safe_path(req.file_path):
|
||||||
|
raise HTTPException(status_code=400, detail="不允许的路径")
|
||||||
path = Path(req.file_path)
|
path = Path(req.file_path)
|
||||||
if not path.exists():
|
if not path.exists():
|
||||||
raise HTTPException(status_code=404, detail=f"文件不存在: {req.file_path}")
|
raise HTTPException(status_code=404, detail=f"文件不存在: {req.file_path}")
|
||||||
|
|||||||
@@ -90,3 +90,23 @@ class TestIngestEndpoint:
|
|||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
data = response.json()
|
data = response.json()
|
||||||
assert len(data["results"]) > 0
|
assert len(data["results"]) > 0
|
||||||
|
|
||||||
|
|
||||||
|
class TestSecurity:
|
||||||
|
"""安全测试."""
|
||||||
|
|
||||||
|
def test_ingest_rejects_path_traversal(self, client):
|
||||||
|
"""拒绝路径遍历攻击."""
|
||||||
|
response = client.post(
|
||||||
|
"/api/v1/ingest",
|
||||||
|
json={"file_path": "../../../etc/passwd"},
|
||||||
|
)
|
||||||
|
assert response.status_code in (400, 403)
|
||||||
|
|
||||||
|
def test_ingest_rejects_absolute_path(self, client):
|
||||||
|
"""拒绝绝对路径."""
|
||||||
|
response = client.post(
|
||||||
|
"/api/v1/ingest",
|
||||||
|
json={"file_path": "C:\\Windows\\System32\\config\\SAM"},
|
||||||
|
)
|
||||||
|
assert response.status_code in (400, 403)
|
||||||
|
|||||||
Reference in New Issue
Block a user