mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-06-17 10:15:27 -04:00
fix(personal): resolve upload delete path (#4291)
This commit is contained in:
@@ -278,8 +278,8 @@ def setup_personal_routes(personal_docs_manager, rag_manager, rag_available):
|
||||
# Delete file from disk if it's in uploads dir
|
||||
deleted_from_disk = False
|
||||
try:
|
||||
abs_target = os.path.abspath(filepath)
|
||||
base_abs = os.path.abspath(UPLOADS_DIR)
|
||||
abs_target = os.path.realpath(filepath)
|
||||
base_abs = os.path.realpath(UPLOADS_DIR)
|
||||
in_uploads = (
|
||||
abs_target == base_abs
|
||||
or os.path.commonpath([abs_target, base_abs]) == base_abs
|
||||
|
||||
@@ -0,0 +1,74 @@
|
||||
import asyncio
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from routes import personal_routes
|
||||
|
||||
|
||||
class _FakePersonalDocs:
|
||||
def __init__(self):
|
||||
self.excluded = []
|
||||
|
||||
def exclude_file(self, filepath):
|
||||
self.excluded.append(filepath)
|
||||
|
||||
|
||||
class _FakeRAG:
|
||||
def __init__(self):
|
||||
self.deleted_sources = []
|
||||
|
||||
def delete_by_source(self, filepath):
|
||||
self.deleted_sources.append(filepath)
|
||||
return 1
|
||||
|
||||
|
||||
def _delete_endpoint(personal_docs):
|
||||
router = personal_routes.setup_personal_routes(personal_docs, None, True)
|
||||
for route in router.routes:
|
||||
if getattr(route, "path", "") == "/api/personal/file" and "DELETE" in getattr(route, "methods", set()):
|
||||
return route.endpoint
|
||||
raise AssertionError("DELETE /api/personal/file endpoint not found")
|
||||
|
||||
|
||||
def test_delete_file_refuses_symlink_directory_escape(tmp_path, monkeypatch):
|
||||
uploads = tmp_path / "uploads"
|
||||
uploads.mkdir()
|
||||
outside = tmp_path / "outside"
|
||||
outside.mkdir()
|
||||
victim = outside / "victim.txt"
|
||||
victim.write_text("keep me", encoding="utf-8")
|
||||
os.symlink(outside, uploads / "linked")
|
||||
|
||||
docs = _FakePersonalDocs()
|
||||
rag = _FakeRAG()
|
||||
monkeypatch.setattr(personal_routes, "UPLOADS_DIR", str(uploads))
|
||||
monkeypatch.setattr(personal_routes, "get_rag_manager", lambda: rag)
|
||||
|
||||
filepath = str(uploads / "linked" / "victim.txt")
|
||||
result = asyncio.run(_delete_endpoint(docs)(filepath=filepath, owner="alice", _admin=None))
|
||||
|
||||
assert result["deleted_from_disk"] is False
|
||||
assert victim.read_text(encoding="utf-8") == "keep me"
|
||||
assert docs.excluded == [filepath]
|
||||
assert rag.deleted_sources == [filepath]
|
||||
|
||||
|
||||
def test_delete_file_removes_regular_file_inside_upload_root(tmp_path, monkeypatch):
|
||||
uploads = tmp_path / "uploads"
|
||||
uploads.mkdir()
|
||||
uploaded_file = uploads / "alice" / "notes.txt"
|
||||
uploaded_file.parent.mkdir()
|
||||
uploaded_file.write_text("delete me", encoding="utf-8")
|
||||
|
||||
docs = _FakePersonalDocs()
|
||||
rag = _FakeRAG()
|
||||
monkeypatch.setattr(personal_routes, "UPLOADS_DIR", str(uploads))
|
||||
monkeypatch.setattr(personal_routes, "get_rag_manager", lambda: rag)
|
||||
|
||||
filepath = str(uploaded_file)
|
||||
result = asyncio.run(_delete_endpoint(docs)(filepath=filepath, owner="alice", _admin=None))
|
||||
|
||||
assert result["deleted_from_disk"] is True
|
||||
assert not uploaded_file.exists()
|
||||
assert docs.excluded == [filepath]
|
||||
assert rag.deleted_sources == [filepath]
|
||||
Reference in New Issue
Block a user