fix(download): check task ownership before serving queued files (#1357)

`/api/localdownload` resolves the file through the live queue and
returns it before checking who owns the task; the owner check only runs
on the download-history fallback, once the task has aged out of the
queue. Task ids are source ids, so two users who searched the same book
can end up with the same id.

This applies the same rule on the queue path, reusing the
`_task_owned_by_actor` helper the cancel/retry/priority routes already
use, so both paths answer a non-owner with the same 404. Admin behaviour
is unchanged.

The 404 matches what this endpoint's history path already returns for a
non-owner rather than the 403 `download_not_owned` the
cancel/retry/priority routes use, happy to switch it if you prefer
consistency with the siblings instead.

## Verification

- `tests/core/test_activity_routes_api.py`: the owner still receives
their queued file; a different user receives 404. The new case fails on
current main and passes here; the existing history-fallback test is
unchanged.
- Full suite (3148), ruff, ruff format, basedpyright, vulture green.
This commit is contained in:
splitsec2
2026-09-20 12:56:23 -04:00
committed by GitHub
parent 7934924678
commit c1315a2b23
2 changed files with 74 additions and 0 deletions
+11
View File
@@ -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)
+63
View File
@@ -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
):