diff --git a/shelfmark/main.py b/shelfmark/main.py index d6c6928b..03fffefd 100644 --- a/shelfmark/main.py +++ b/shelfmark/main.py @@ -2846,7 +2846,10 @@ def api_metadata_search() -> Response | tuple[Response, int]: # Extract custom search field values from query params fields: dict[str, Any] = {} + may_use_account = _session_may_use_provider_account() for search_field in provider.search_fields: + if search_field.key in provider.account_search_fields and not may_use_account: + continue value = request.args.get(search_field.key) if value is not None: # Strip string values to handle whitespace-only input @@ -2935,6 +2938,9 @@ def api_metadata_field_options() -> Response: if not provider or not provider.is_available(): return jsonify({"options": []}) + if field_key in provider.account_search_fields and not _session_may_use_provider_account(): + return jsonify({"options": []}) + options = provider.get_search_field_options(field_key, query=query_text or None) return jsonify({"options": options}) except _OPERATIONAL_ERRORS as e: @@ -2951,6 +2957,16 @@ def _attach_library_ownership(book: Any, book_dict: dict[str, Any]) -> None: book_dict["library"] = owned +def _session_may_use_provider_account() -> bool: + """Whether this session may read or change the metadata provider's own account. + + A provider such as Hardcover is signed in as one account, the instance owner's, and its lists + and shelves belong to that person. With no authentication configured the one user is the owner. + Otherwise only an admin may use them. + """ + return get_auth_mode() == "none" or bool(session.get("is_admin", False)) + + def _resolve_metadata_provider(provider_name: str) -> MetadataProvider: """Validate, instantiate and return a ready metadata provider. @@ -3051,6 +3067,8 @@ def _handle_target_errors( def api_metadata_book_targets(provider: str, book_id: str) -> Response | tuple[Response, int]: """Get provider-managed list/status targets for a specific book.""" prov = _resolve_metadata_provider(provider) + if not _session_may_use_provider_account(): + return jsonify({"options": []}) return jsonify({"options": prov.get_book_targets(book_id)}) @@ -3068,6 +3086,8 @@ def api_metadata_book_targets_batch(provider: str) -> Response | tuple[Response, book_ids = [str(bid) for bid in raw_ids[:50]] + if not _session_may_use_provider_account(): + return jsonify({"results": {book_id: [] for book_id in book_ids}}) return jsonify({"results": prov.get_book_targets_batch(book_ids)}) @@ -3078,6 +3098,8 @@ def api_metadata_book_targets_update( provider: str, book_id: str ) -> Response | tuple[Response, int]: """Set whether a book belongs to a provider-managed list or shelf.""" + if not _session_may_use_provider_account(): + return jsonify({"error": "Admin access required"}), 403 prov = _resolve_metadata_provider(provider) payload = request.get_json(silent=True) or {} diff --git a/shelfmark/metadata_providers/__init__.py b/shelfmark/metadata_providers/__init__.py index 8a856fb3..77aca6f9 100644 --- a/shelfmark/metadata_providers/__init__.py +++ b/shelfmark/metadata_providers/__init__.py @@ -347,6 +347,7 @@ class MetadataProvider(ABC): requires_auth: True if API key/authentication is required supported_sorts: List of SortOrder values this provider supports search_fields: List of provider-specific search fields + account_search_fields: Search fields that read the provider's own account, admin only capabilities: Declarative capabilities exposed to shared UI code """ @@ -356,6 +357,9 @@ class MetadataProvider(ABC): requires_auth: bool supported_sorts: ClassVar[tuple[SortOrder, ...]] = (SortOrder.RELEVANCE,) search_fields: ClassVar[tuple[SearchField, ...]] = () + # Search fields that read the provider's own account (private lists, shelves) rather than public + # data. Only an admin may use them; see ``_session_may_use_provider_account``. + account_search_fields: ClassVar[frozenset[str]] = frozenset() capabilities: ClassVar[tuple[MetadataCapability, ...]] = () @abstractmethod diff --git a/shelfmark/metadata_providers/hardcover.py b/shelfmark/metadata_providers/hardcover.py index 9a817793..888c5ef2 100644 --- a/shelfmark/metadata_providers/hardcover.py +++ b/shelfmark/metadata_providers/hardcover.py @@ -1097,6 +1097,7 @@ class HardcoverProvider(MetadataProvider): sort=SortOrder.SERIES_ORDER, ), ) + account_search_fields: ClassVar[frozenset[str]] = frozenset({"hardcover_list"}) search_fields: ClassVar[tuple[SearchField, ...]] = ( TextSearchField( key="author", diff --git a/tests/core/test_provider_account_access.py b/tests/core/test_provider_account_access.py new file mode 100644 index 00000000..4818e05c --- /dev/null +++ b/tests/core/test_provider_account_access.py @@ -0,0 +1,176 @@ +"""Only an admin may use the metadata provider's own account (Hardcover lists and shelves).""" + +from __future__ import annotations + +import importlib +from unittest.mock import patch + +import pytest + +from shelfmark.metadata_providers.hardcover import HardcoverProvider + + +@pytest.fixture(scope="module") +def main_module(): + """Import `shelfmark.main` with background startup disabled.""" + with patch("shelfmark.download.orchestrator.start"): + import shelfmark.main as main + + importlib.reload(main) + return main + + +@pytest.fixture +def provider(monkeypatch): + instance = HardcoverProvider(api_key="test-token") + calls: dict[str, list] = {"searched": [], "targets": [], "batch": [], "set": [], "options": []} + + def _search_paginated(options): + from shelfmark.metadata_providers import SearchResult + + calls["searched"].append(options) + return SearchResult(books=[], page=options.page, total_found=0, has_more=False) + + monkeypatch.setattr(instance, "search_paginated", _search_paginated) + monkeypatch.setattr(instance, "is_available", lambda: True) + monkeypatch.setattr( + instance, + "get_book_targets", + lambda book_id: calls["targets"].append(book_id) or [{"value": "list:1", "label": "Mine"}], + ) + monkeypatch.setattr( + instance, + "get_book_targets_batch", + lambda ids: calls["batch"].append(ids) or {i: [{"value": "list:1"}] for i in ids}, + ) + monkeypatch.setattr( + instance, + "set_book_target_state", + lambda book_id, target, *, selected: ( + calls["set"].append((book_id, target, selected)) or {"changed": True} + ), + ) + monkeypatch.setattr( + instance, + "get_search_field_options", + lambda field_key, query=None: ( + calls["options"].append(field_key) or [{"value": "status:1", "label": "Want to Read"}] + ), + ) + instance.calls = calls # type: ignore[attr-defined] + return instance + + +@pytest.fixture +def client(main_module, provider): + with ( + patch("shelfmark.metadata_providers.get_provider", return_value=provider), + patch("shelfmark.metadata_providers.is_provider_registered", return_value=True), + patch("shelfmark.metadata_providers.is_provider_enabled", return_value=True), + patch("shelfmark.metadata_providers.get_provider_kwargs", return_value={}), + ): + yield main_module.app.test_client() + + +def _session(client, *, is_admin: bool) -> None: + with client.session_transaction() as sess: + sess["user_id"] = "someone" + sess["is_admin"] = is_admin + + +def _as(main_module, client, *, is_admin: bool, auth_mode: str = "builtin"): + _session(client, is_admin=is_admin) + return patch.object(main_module, "get_auth_mode", return_value=auth_mode) + + +class TestTargets: + def test_a_user_sees_no_targets_and_the_account_is_not_asked( + self, main_module, client, provider + ) -> None: + with _as(main_module, client, is_admin=False): + single = client.get("/api/metadata/book/hardcover/42/targets") + batch = client.post( + "/api/metadata/book/hardcover/targets/batch", json={"book_ids": [1, 2]} + ) + + assert single.json == {"options": []} + assert batch.json == {"results": {"1": [], "2": []}} + assert provider.calls["targets"] == [] + assert provider.calls["batch"] == [] + + def test_a_user_cannot_change_a_list(self, main_module, client, provider) -> None: + with _as(main_module, client, is_admin=False): + resp = client.put( + "/api/metadata/book/hardcover/42/targets", + json={"target": "list:1", "selected": True}, + ) + + assert resp.status_code == 403 + assert provider.calls["set"] == [] + + def test_an_admin_keeps_targets(self, main_module, client, provider) -> None: + with _as(main_module, client, is_admin=True): + single = client.get("/api/metadata/book/hardcover/42/targets") + put = client.put( + "/api/metadata/book/hardcover/42/targets", + json={"target": "list:1", "selected": True}, + ) + + assert single.json == {"options": [{"value": "list:1", "label": "Mine"}]} + assert put.status_code == 200 + assert provider.calls["set"] == [("42", "list:1", True)] + + def test_with_no_authentication_the_one_user_keeps_targets( + self, main_module, client, provider + ) -> None: + with _as(main_module, client, is_admin=False, auth_mode="none"): + single = client.get("/api/metadata/book/hardcover/42/targets") + + assert single.json == {"options": [{"value": "list:1", "label": "Mine"}]} + + +class TestListField: + def test_a_user_gets_no_list_options(self, main_module, client, provider) -> None: + with _as(main_module, client, is_admin=False): + resp = client.get("/api/metadata/field-options?provider=hardcover&field=hardcover_list") + + assert resp.json == {"options": []} + assert provider.calls["options"] == [] + + def test_a_user_still_gets_public_suggestions(self, main_module, client, provider) -> None: + with _as(main_module, client, is_admin=False): + resp = client.get("/api/metadata/field-options?provider=hardcover&field=author&query=x") + + assert resp.json["options"] + assert provider.calls["options"] == ["author"] + + def test_an_admin_gets_the_list_options(self, main_module, client, provider) -> None: + with _as(main_module, client, is_admin=True): + resp = client.get("/api/metadata/field-options?provider=hardcover&field=hardcover_list") + + assert resp.json["options"] + + def test_a_user_cannot_browse_a_list_by_asking_for_it( + self, main_module, client, provider + ) -> None: + with _as(main_module, client, is_admin=False): + client.get("/api/metadata/search?provider=hardcover&hardcover_list=status:1&query=dune") + + (options,) = provider.calls["searched"] + assert "hardcover_list" not in options.fields + + def test_a_user_can_still_search_with_public_fields( + self, main_module, client, provider + ) -> None: + with _as(main_module, client, is_admin=False): + client.get("/api/metadata/search?provider=hardcover&author=id:1&query=dune") + + (options,) = provider.calls["searched"] + assert options.fields == {"author": "id:1"} + + def test_an_admin_can_browse_a_list(self, main_module, client, provider) -> None: + with _as(main_module, client, is_admin=True): + client.get("/api/metadata/search?provider=hardcover&hardcover_list=status:1&query=dune") + + (options,) = provider.calls["searched"] + assert options.fields["hardcover_list"] == "status:1"