diff --git a/routes/personal_routes.py b/routes/personal_routes.py index e6906223e..a42615be7 100644 --- a/routes/personal_routes.py +++ b/routes/personal_routes.py @@ -358,11 +358,13 @@ def setup_personal_routes(personal_docs_manager, rag_manager, rag_available): except Exception as e: logger.warning(f"RAG removal failed for {filepath}: {e}") - # Delete file from disk if it's in uploads dir + # Delete file from disk if it's in the caller's own uploads dir. + # Scope to the per-owner subdir, not the shared uploads root, so one + # admin can't delete another user's personal files by path. deleted_from_disk = False try: abs_target = os.path.realpath(filepath) - base_abs = os.path.realpath(UPLOADS_DIR) + base_abs = os.path.realpath(_personal_upload_dir_for_owner(owner, create=False)) in_uploads = ( abs_target == base_abs or os.path.commonpath([abs_target, base_abs]) == base_abs diff --git a/tests/test_personal_delete_file_confinement.py b/tests/test_personal_delete_file_confinement.py index 346a15028..03bc40db0 100644 --- a/tests/test_personal_delete_file_confinement.py +++ b/tests/test_personal_delete_file_confinement.py @@ -72,3 +72,24 @@ def test_delete_file_removes_regular_file_inside_upload_root(tmp_path, monkeypat assert not uploaded_file.exists() assert docs.excluded == [filepath] assert rag.deleted_sources == [filepath] + + +def test_delete_file_refuses_other_owners_upload(tmp_path, monkeypatch): + # alice must not be able to delete a file living under bob's per-owner + # upload subdir, even though it sits inside the shared uploads root. + uploads = tmp_path / "uploads" + uploads.mkdir() + victim = uploads / "bob" / "secret.txt" + victim.parent.mkdir() + victim.write_text("keep 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(victim) + 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"