diff --git a/shelfmark/main.py b/shelfmark/main.py index da84d54e..83fdbd2b 100644 --- a/shelfmark/main.py +++ b/shelfmark/main.py @@ -1649,6 +1649,17 @@ def api_local_download() -> Response | tuple[Response, int]: # Book data not found or not available return jsonify({"error": "File not found"}), 404 + + is_admin, db_user_id, can_access_status = _resolve_status_scope() + if not is_admin: + actor_username = session.get("user_id") + if not can_access_status or not _task_owned_by_actor( + book_info, + actor_user_id=db_user_id, + actor_username=actor_username if isinstance(actor_username, str) else None, + ): + return jsonify({"error": "File not found"}), 404 + file_name = book_info.get_filename() if book_info is not None else Path(book_id).name # Prepare the file for sending to the client data = io.BytesIO(file_data) diff --git a/tests/core/test_activity_routes_api.py b/tests/core/test_activity_routes_api.py index ac6371d8..f65b85b4 100644 --- a/tests/core/test_activity_routes_api.py +++ b/tests/core/test_activity_routes_api.py @@ -10,6 +10,8 @@ from unittest.mock import ANY, patch import pytest +from shelfmark.core.models import DownloadTask, QueueStatus + @pytest.fixture(scope="module") def main_module(): @@ -321,6 +323,67 @@ class TestActivityRoutes: assert response.data == file_bytes assert "attachment" in response.headers.get("Content-Disposition", "").lower() + def test_localdownload_serves_own_live_queue_task(self, main_module, client, tmp_path): + user = _create_user(main_module, prefix="reader") + _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) + + task_id = f"live-localdownload-{uuid.uuid4().hex[:8]}" + file_path = tmp_path / "live-owned.epub" + file_bytes = b"live download payload" + file_path.write_bytes(file_bytes) + + task = DownloadTask( + task_id=task_id, + source="direct_download", + title="Live Local Download", + user_id=user["id"], + username=user["username"], + download_path=str(file_path), + ) + assert main_module.backend.book_queue.add(task) is True + + try: + main_module.backend.book_queue.update_status(task_id, QueueStatus.COMPLETE) + with patch.object(main_module, "get_auth_mode", return_value="builtin"): + response = client.get(f"/api/localdownload?id={task_id}") + finally: + main_module.backend.book_queue.cancel_download(task_id) + + assert response.status_code == 200 + assert response.data == file_bytes + + def test_localdownload_returns_not_found_for_other_users_live_queue_task( + self, main_module, client, tmp_path + ): + owner = _create_user(main_module, prefix="owner") + viewer = _create_user(main_module, prefix="reader") + _set_session(client, user_id=viewer["username"], db_user_id=viewer["id"], is_admin=False) + + task_id = f"live-localdownload-{uuid.uuid4().hex[:8]}" + file_path = tmp_path / "live-other.epub" + file_bytes = b"live download payload" + file_path.write_bytes(file_bytes) + + task = DownloadTask( + task_id=task_id, + source="direct_download", + title="Live Local Download", + user_id=owner["id"], + username=owner["username"], + download_path=str(file_path), + ) + assert main_module.backend.book_queue.add(task) is True + + try: + main_module.backend.book_queue.update_status(task_id, QueueStatus.COMPLETE) + with patch.object(main_module, "get_auth_mode", return_value="builtin"): + response = client.get(f"/api/localdownload?id={task_id}") + finally: + main_module.backend.book_queue.cancel_download(task_id) + + assert response.status_code == 404 + assert response.json == {"error": "File not found"} + def test_dismiss_legacy_fulfilled_request_creates_minimal_history_snapshot( self, main_module, client ):