starred/shelfmark
1
0
Fork 0
mirror of https://github.com/calibrain/shelfmark.git synced 2026-10-05 17:31:06 +01:00
Code Issues Packages Projects Releases 10 Wiki Activity

Validate request policy release source (#975)

Browse Source
This commit is contained in:
Alex
2026-05-10 09:46:31 +01:00
committed by GitHub
parent f6357ead41
commit 0e120abfaf
2 changed files with 188 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shelfmark/core/request_routes.py
+28
View File
@@ -220,6 +220,26 @@ def _normalize_release_result_request_payload(
return "release", normalized_release_data
def _validate_release_source_matches_policy_context(
*,
source: str,
release_data: object,
) -> None:
if not isinstance(release_data, dict):
return
release_source = normalize_source(release_data.get("source"))
if release_source in {"", "*"} or release_source == source:
return
msg = "Policy context source must match release_data.source"
raise RequestServiceError(
msg,
status_code=400,
code="policy_source_mismatch",
)
def _resolve_request_title(request_row: dict[str, Any]) -> str:
return _resolve_title_from_book_data(request_row.get("book_data"))
@@ -317,6 +337,10 @@ def _prepare_request_create_arguments(
content_type = normalize_content_type(
context.get("content_type") or data.get("content_type") or book_data.get("content_type")
)
_validate_release_source_matches_policy_context(
source=source,
release_data=release_data,
)
request_level, release_data = _normalize_release_result_request_payload(
source=source,
request_level=request_level,
@@ -324,6 +348,10 @@ def _prepare_request_create_arguments(
release_data=release_data,
content_type=content_type,
)
_validate_release_source_matches_policy_context(
source=source,
release_data=release_data,
)
global_settings, user_settings, effective, requests_enabled = _resolve_effective_policy(
user_db,
tests/core/test_request_routes_api.py
+160
View File
@@ -354,6 +354,166 @@ class TestRequestRoutes:
mock_notify_admin.assert_not_called()
mock_notify_user.assert_not_called()
def test_download_policy_rejects_mismatched_context_and_release_source(
self, main_module, client
):
user = _create_user(main_module, prefix="reader")
_set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False)
policy = _policy(
default_ebook="download",
rules=[{"source": "direct_download", "content_type": "*", "mode": "blocked"}],
)
payload = {
"book_data": {
"title": "Policy Source Mismatch",
"author": "Shelfmark",
"content_type": "ebook",
"provider": "openlibrary",
"provider_id": "policy-source-mismatch-1",
},
"context": {
"source": "prowlarr",
"content_type": "ebook",
"request_level": "release",
},
"release_data": {
"source": "direct_download",
"source_id": "blocked-release-1",
"title": "Blocked Release.epub",
},
}
with patch.object(main_module, "get_auth_mode", return_value="builtin"):
with patch.object(
main_module, "load_users_request_policy_settings", return_value=policy
):
with patch(
"shelfmark.core.request_routes.load_users_request_policy_settings",
return_value=policy,
):
with patch.object(main_module.backend, "queue_release") as mock_queue:
resp = client.post("/api/requests", json=payload)
assert resp.status_code == 400
assert resp.json["code"] == "policy_source_mismatch"
assert resp.json["error"] == "Policy context source must match release_data.source"
assert main_module.user_db.list_requests(user_id=user["id"]) == []
mock_queue.assert_not_called()
def test_release_result_source_rejects_mismatch_before_normalization(self, main_module, client):
user = _create_user(main_module, prefix="reader")
_set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False)
policy = _policy(
default_ebook="download",
rules=[{"source": "prowlarr", "content_type": "*", "mode": "blocked"}],
)
payload = {
"book_data": {
"title": "Release Result Source Mismatch",
"author": "Shelfmark",
"content_type": "ebook",
"provider": "openlibrary",
"provider_id": "release-result-mismatch-1",
},
"context": {
"source": "direct_download",
"content_type": "ebook",
"request_level": "release",
},
"release_data": {
"source": "prowlarr",
"source_id": "blocked-prowlarr-release-1",
"title": "Blocked Prowlarr Release.epub",
},
}
with patch.object(main_module, "get_auth_mode", return_value="builtin"):
with patch.object(
main_module, "load_users_request_policy_settings", return_value=policy
):
with patch(
"shelfmark.core.request_routes.load_users_request_policy_settings",
return_value=policy,
):
with patch.object(main_module.backend, "queue_release") as mock_queue:
resp = client.post("/api/requests", json=payload)
assert resp.status_code == 400
assert resp.json["code"] == "policy_source_mismatch"
assert resp.json["error"] == "Policy context source must match release_data.source"
assert main_module.user_db.list_requests(user_id=user["id"]) == []
mock_queue.assert_not_called()
def test_batch_rejects_release_result_source_mismatch_before_creating_any_requests(
self, main_module, client
):
user = _create_user(main_module, prefix="reader")
_set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False)
policy = _policy(
default_ebook="request_release",
rules=[{"source": "prowlarr", "content_type": "*", "mode": "blocked"}],
)
payloads = [
{
"book_data": {
"title": "Batch Valid Direct",
"author": "Shelfmark",
"content_type": "ebook",
"provider": "openlibrary",
"provider_id": "batch-valid-direct-1",
},
"context": {
"source": "direct_download",
"content_type": "ebook",
"request_level": "release",
},
"release_data": {
"source": "direct_download",
"source_id": "batch-valid-direct-release-1",
"title": "Batch Valid Direct.epub",
},
},
{
"book_data": {
"title": "Batch Release Result Mismatch",
"author": "Shelfmark",
"content_type": "ebook",
"provider": "openlibrary",
"provider_id": "batch-release-result-mismatch-1",
},
"context": {
"source": "direct_download",
"content_type": "ebook",
"request_level": "release",
},
"release_data": {
"source": "prowlarr",
"source_id": "batch-blocked-prowlarr-release-1",
"title": "Batch Blocked Prowlarr.epub",
},
},
]
with patch.object(main_module, "get_auth_mode", return_value="builtin"):
with patch.object(
main_module, "load_users_request_policy_settings", return_value=policy
):
with patch(
"shelfmark.core.request_routes.load_users_request_policy_settings",
return_value=policy,
):
with patch.object(main_module.backend, "queue_release") as mock_queue:
resp = client.post("/api/requests/batch", json={"requests": payloads})
assert resp.status_code == 400
assert resp.json["code"] == "policy_source_mismatch"
assert resp.json["error"] == "Policy context source must match release_data.source"
assert main_module.user_db.list_requests(user_id=user["id"]) == []
mock_queue.assert_not_called()
def test_batch_download_policy_queues_releases_without_creating_requests(
self, main_module, client
):