diff --git a/.agent-work/ledger.json b/.agent-work/ledger.json index 84d6e0c..8240ce2 100644 --- a/.agent-work/ledger.json +++ b/.agent-work/ledger.json @@ -4,6 +4,11 @@ "events": [ "Created tools package (tools/__init__.py, tools/example.py), examples/input.json, tests/test_example.py, .gitignore, .env.example; updated pyproject.toml (metadata, setuptools packages, dev pytest extra, pytest pythonpath/testpaths, requires-python >=3.11 due to approved-mirror exceptiongroup constraint) and generated uv.lock via prepare_environment update. Facade validate/test/smoke and --help all pass locally." ] + }, + "projects/product-54354138-1#4": { + "events": [ + "Implemented tools/errors.py, tools/sources.py and tests/test_sources.py for the credential-free source registry and HTTPS fetch layer. Verified with the facade commands tools.validate, tools.test (54 passed) and tools.smoke, all green on the integrated branch (target task/1-task was already at HEAD). No protected facade paths touched." + ] } } } diff --git a/tests/test_sources.py b/tests/test_sources.py new file mode 100644 index 0000000..cf02324 --- /dev/null +++ b/tests/test_sources.py @@ -0,0 +1,452 @@ +"""Tests for the credential-free image source registry and HTTPS layer.""" + +from __future__ import annotations + +import urllib.parse + +import pytest + +from tools import sources +from tools.errors import ( + DuplicateSourceError, + HostNotAllowedError, + SearchRequestError, + SourceAccessError, + SourceHTTPError, + SourceUnreachableError, + ToolkitError, + UnknownSourceError, +) +from tools.sources import ( + APPROVED_HOSTS, + DEFAULT_LIMIT, + HTTPSClient, + HTTPResponse, + OpenverseAdapter, + SourceRegistry, + WikimediaCommonsAdapter, + build_registry, + is_approved_url, + validate_source_url, +) + + +urllib_parse = urllib.parse.urlsplit +urllib_parse_qsl = urllib.parse.parse_qsl + + +class FakeTransport: + """Scripted transport: records requests and replays canned responses.""" + + def __init__(self, responses): + self._responses = list(responses) + self.calls = [] + + def __call__(self, url, headers, timeout): + self.calls.append({"url": url, "headers": dict(headers), "timeout": timeout}) + item = self._responses.pop(0) if len(self._responses) > 1 else self._responses[0] + if isinstance(item, Exception): + raise item + return item + + +def json_response(payload, status=200): + import json + + return HTTPResponse(status=status, body=json.dumps(payload).encode("utf-8")) + + +def make_client(transport, **kwargs): + kwargs.setdefault("sleep", lambda _seconds: None) + kwargs.setdefault("max_attempts", 3) + return HTTPSClient(transport=transport, **kwargs) + + +# -- registry --------------------------------------------------------------- + + +def test_default_registry_reports_available_source_ids(): + registry = build_registry() + assert registry.available_sources() == ["openverse", "wikimedia_commons"] + assert "wikimedia_commons" in registry + assert "openverse" in registry + assert len(registry) == 2 + assert sorted(iter(registry)) == ["openverse", "wikimedia_commons"] + + +def test_registry_lookup_returns_registered_adapter(): + registry = build_registry() + assert isinstance(registry.get("wikimedia_commons"), WikimediaCommonsAdapter) + assert isinstance(registry.get("openverse"), OpenverseAdapter) + assert registry.get("openverse").source_id == "openverse" + + +def test_registry_rejects_duplicate_registration(): + registry = build_registry() + with pytest.raises(DuplicateSourceError) as excinfo: + registry.register(OpenverseAdapter()) + assert excinfo.value.source_id == "openverse" + assert isinstance(excinfo.value, ToolkitError) + # replacing explicitly is still possible + registry.register(OpenverseAdapter(), replace=True) + assert registry.available_sources() == ["openverse", "wikimedia_commons"] + + +def test_unknown_source_raises_typed_error_listing_available_ids(): + registry = build_registry() + with pytest.raises(UnknownSourceError) as excinfo: + registry.get("nope") + error = excinfo.value + assert error.available == ("openverse", "wikimedia_commons") + assert "openverse" in str(error) + assert "wikimedia_commons" in str(error) + assert error.details["available_sources"] == ["openverse", "wikimedia_commons"] + assert error.code == "source_unknown" + + +def test_registry_search_with_unknown_source_does_not_contact_network(): + transport = FakeTransport([RuntimeError("must not be called")]) + registry = build_registry(client=make_client(transport)) + with pytest.raises(UnknownSourceError): + registry.search("missing", "cats") + assert transport.calls == [] + + +# -- request construction --------------------------------------------------- + +COMMONS_PAYLOAD = { + "query": { + "pages": [ + { + "title": "File:Cat one.svg", + "imageinfo": [ + { + "url": "https://upload.wikimedia.org/wikipedia/commons/1/1a/Cat_one.svg", + "mime": "image/svg+xml", + "width": 48, + "height": 48, + "descriptionurl": "https://commons.wikimedia.org/wiki/File:Cat_one.svg", + } + ], + }, + { + "title": "File:Cat two.png", + "imageinfo": [ + { + "url": "https://upload.wikimedia.org/wikipedia/commons/2/2b/Cat_two.png", + "mime": "image/png", + "width": 64, + "height": 64, + "descriptionurl": "https://commons.wikimedia.org/wiki/File:Cat_two.png", + } + ], + }, + { + "title": "File:Cat three.ico", + "imageinfo": [ + { + "url": "https://upload.wikimedia.org/wikipedia/commons/3/3c/Cat_three.ico", + "mime": "image/x-icon", + "width": 32, + "height": 32, + "descriptionurl": "https://commons.wikimedia.org/wiki/File:Cat_three.ico", + } + ], + }, + ] + } +} + +OPENVERSE_PAYLOAD = { + "result_count": 2, + "results": [ + { + "title": "Sleeping cat", + "url": "https://live.staticflickr.com/1/cat.jpg", + "thumbnail": "https://api.openverse.org/v1/images/aaa/thumb/", + "filetype": "jpg", + "width": 1024, + "height": 768, + "license": "by", + "creator": "someone", + "foreign_landing_url": "https://www.flickr.com/photos/someone/1", + }, + { + "title": "Second cat", + "url": "https://live.staticflickr.com/2/cat.jpg", + "thumbnail": "https://api.openverse.org/v1/images/bbb/thumb/", + "filetype": "jpg", + "width": 800, + "height": 600, + "license": "cc0", + "creator": "other", + "foreign_landing_url": "https://www.flickr.com/photos/other/2", + }, + ], +} + + +def test_commons_search_url_carries_topic_and_limit(): + adapter = WikimediaCommonsAdapter(client=make_client(FakeTransport([HTTPResponse(500, b"")]))) + url = adapter.build_search_url("sleeping cat", 3) + parsed = urllib_parse(url) + assert parsed.scheme == "https" + assert parsed.hostname == "commons.wikimedia.org" + assert parsed.port is None + query = dict(urllib_parse_qsl(parsed.query)) + assert query["gsrsearch"] == "sleeping cat" + assert query["gsrlimit"] == "3" + assert query["action"] == "query" + assert query["format"] == "json" + assert query["generator"] == "search" + assert query["gsrnamespace"] == "6" + assert "imageinfo" in query["prop"] + assert is_approved_url(url) + + +def test_openverse_search_url_carries_topic_and_limit(): + adapter = OpenverseAdapter(client=make_client(FakeTransport([HTTPResponse(500, b"")]))) + url = adapter.build_search_url("sleeping cat", 4) + parsed = urllib_parse(url) + assert parsed.scheme == "https" + assert parsed.hostname == "api.openverse.org" + assert parsed.path == "/v1/images/" + query = dict(urllib_parse_qsl(parsed.query)) + assert query["q"] == "sleeping cat" + assert query["page_size"] == "4" + assert is_approved_url(url) + + +def test_topic_and_limit_reach_the_request_through_the_registry(): + transport = FakeTransport([json_response(COMMONS_PAYLOAD)]) + registry = build_registry(client=make_client(transport)) + registry.search("wikimedia_commons", " red panda ", limit=2) + assert len(transport.calls) == 1 + query = dict(urllib_parse_qsl(urllib_parse(transport.calls[0]["url"]).query)) + assert query["gsrsearch"] == "red panda" + assert query["gsrlimit"] == "2" + + +@pytest.mark.parametrize("topic,limit", [("", 3), (" ", 3), ("cats", 0), ("cats", -1), ("cats", 10_000)]) +def test_invalid_topic_or_limit_is_rejected_before_any_request(topic, limit): + transport = FakeTransport([json_response(COMMONS_PAYLOAD)]) + registry = build_registry(client=make_client(transport)) + with pytest.raises(SearchRequestError): + registry.search("wikimedia_commons", topic, limit=limit) + assert transport.calls == [] + + +def test_default_limit_is_used_when_caller_omits_it(): + transport = FakeTransport([json_response(COMMONS_PAYLOAD)]) + registry = build_registry(client=make_client(transport)) + registry.search("wikimedia_commons", "cats") + query = dict(urllib_parse_qsl(urllib_parse(transport.calls[0]["url"]).query)) + assert query["gsrlimit"] == str(DEFAULT_LIMIT) + + +# -- normalization ---------------------------------------------------------- + + +def test_commons_results_are_normalized_into_candidates(): + transport = FakeTransport([json_response(COMMONS_PAYLOAD)]) + adapter = WikimediaCommonsAdapter(client=make_client(transport)) + candidates = adapter.search("cats", 3) + assert len(candidates) == 3 + first = candidates[0] + assert first.source_id == "wikimedia_commons" + assert first.file_url.endswith("Cat_one.svg") + assert first.mime_type == "image/svg+xml" + assert first.width == 48 and first.height == 48 + assert first.page_url.startswith("https://commons.wikimedia.org/") + assert callable(first.fetch_bytes) + assert first.to_dict()["mime_type"] == "image/svg+xml" + assert "fetch_bytes" not in first.to_dict() + + +def test_commons_respects_the_requested_limit(): + transport = FakeTransport([json_response(COMMONS_PAYLOAD)]) + adapter = WikimediaCommonsAdapter(client=make_client(transport)) + assert len(adapter.search("cats", 1)) == 1 + assert len(adapter.search("cats", 2)) == 2 + + +def test_openverse_results_are_normalized_into_candidates(): + transport = FakeTransport([json_response(OPENVERSE_PAYLOAD)]) + adapter = OpenverseAdapter(client=make_client(transport)) + candidates = adapter.search("cats", 5) + assert len(candidates) == 2 + first = candidates[0] + assert first.source_id == "openverse" + assert first.file_url == "https://api.openverse.org/v1/images/aaa/thumb/" + assert first.mime_type == "jpg" + assert first.extra["original_url"] == "https://live.staticflickr.com/1/cat.jpg" + assert callable(first.fetch_bytes) + + +def test_candidate_fetch_bytes_downloads_through_the_client(): + transport = FakeTransport([json_response(COMMONS_PAYLOAD), HTTPResponse(200, b"IMG-bytes")]) + adapter = WikimediaCommonsAdapter(client=make_client(transport)) + candidate = adapter.search("cats", 1)[0] + assert candidate.fetch_bytes() == b"IMG-bytes" + assert transport.calls[-1]["url"] == candidate.file_url + assert transport.calls[-1]["headers"]["User-Agent"] == sources.USER_AGENT + + +def test_adapters_send_an_identifying_user_agent(): + transport = FakeTransport([json_response(OPENVERSE_PAYLOAD)]) + OpenverseAdapter(client=make_client(transport)).search("cats", 1) + agent = transport.calls[0]["headers"]["User-Agent"] + assert sources.PRODUCT_NAME in agent + assert "toolkit" in agent.lower() + + +# -- retry / backoff -------------------------------------------------------- + + +def test_retry_statuses_are_retried_then_succeed(): + sleeps = [] + transport = FakeTransport( + [HTTPResponse(429, b""), HTTPResponse(503, b""), json_response(OPENVERSE_PAYLOAD)] + ) + client = HTTPSClient(transport=transport, sleep=sleeps.append, max_attempts=3) + response = client.get("https://api.openverse.org/v1/images/?q=cats", source_id="openverse") + assert response.status == 200 + assert len(transport.calls) == 3 + assert sleeps == [client.backoff_delay(1), client.backoff_delay(2)] + assert sleeps[0] < sleeps[1] + + +def test_exhausted_retries_raise_source_http_error(): + transport = FakeTransport([HTTPResponse(500, b"boom")]) + sleeps = [] + client = HTTPSClient(transport=transport, sleep=sleeps.append, max_attempts=3) + with pytest.raises(SourceHTTPError) as excinfo: + client.get("https://api.openverse.org/v1/images/?q=cats", source_id="openverse") + assert excinfo.value.status == 500 + assert len(transport.calls) == 3 + assert len(sleeps) == 2 + assert isinstance(excinfo.value, SourceAccessError) + + +def test_backoff_is_capped(): + client = HTTPSClient(transport=FakeTransport([HTTPResponse(500, b"")]), sleep=lambda _s: None) + assert client.backoff_delay(50) <= sources.MAX_BACKOFF_SECONDS + + +def test_non_retryable_status_is_raised_immediately(): + transport = FakeTransport([HTTPResponse(404, b"nope")]) + client = HTTPSClient(transport=transport, sleep=lambda _s: None, max_attempts=3) + with pytest.raises(SourceHTTPError): + client.get("https://commons.wikimedia.org/w/api.php?action=query") + assert len(transport.calls) == 1 + + +def test_unreachable_host_raises_access_error_not_empty_result(): + transport = FakeTransport([OSError("name resolution failed")]) + client = HTTPSClient(transport=transport, sleep=lambda _s: None, max_attempts=2) + with pytest.raises(SourceUnreachableError) as excinfo: + client.get("https://commons.wikimedia.org/w/api.php?action=query") + assert len(transport.calls) == 2 + assert isinstance(excinfo.value, SourceAccessError) + + +def test_search_propagates_access_errors_instead_of_returning_empty_list(): + transport = FakeTransport([OSError("network down")]) + registry = build_registry(client=make_client(transport, max_attempts=1)) + with pytest.raises(SourceAccessError): + registry.search("wikimedia_commons", "cats") + + +def test_malformed_payload_raises_access_error(): + transport = FakeTransport([HTTPResponse(200, b"not json")]) + adapter = OpenverseAdapter(client=make_client(transport)) + with pytest.raises(SourceAccessError): + adapter.search("cats", 1) + + +# -- host allowlist --------------------------------------------------------- + +REJECTED_URLS = [ + "http://commons.wikimedia.org/w/api.php?action=query", + "https://evil.example.com/cat.png", + "https://commons.wikimedia.org.evil.example/w/api.php", + "file:///etc/passwd", + "ftp://upload.wikimedia.org/cat.png", + "https://commons.wikimedia.org:8443/w/api.php", + "", +] + + +@pytest.mark.parametrize("url", REJECTED_URLS) +def test_host_allowlist_rejects_unapproved_urls_without_network(url): + transport = FakeTransport([RuntimeError("network must not be contacted")]) + client = HTTPSClient(transport=transport, sleep=lambda _s: None) + with pytest.raises(HostNotAllowedError): + client.get(url) + assert transport.calls == [] + assert is_approved_url(url) is False + + +@pytest.mark.parametrize( + "url", + [ + "https://commons.wikimedia.org/w/api.php?action=query", + "https://upload.wikimedia.org/wikipedia/commons/1/1a/Cat.svg", + "https://api.openverse.org/v1/images/?q=cats", + "https://commons.wikimedia.org:443/w/api.php", + ], +) +def test_host_allowlist_accepts_approved_https_hosts(url): + assert validate_source_url(url) == url + assert is_approved_url(url) is True + + +def test_approved_hosts_are_the_two_credential_free_sources(): + assert "commons.wikimedia.org" in APPROVED_HOSTS + assert "upload.wikimedia.org" in APPROVED_HOSTS + assert "api.openverse.org" in APPROVED_HOSTS + + +# -- credential-free guarantee --------------------------------------------- + + +def test_search_takes_no_credential_arguments(): + import inspect + + for adapter in (WikimediaCommonsAdapter, OpenverseAdapter): + params = set(inspect.signature(adapter.search).parameters) - {"self"} + assert params == {"topic", "limit"}, params + + registry = build_registry() + params = set(inspect.signature(registry.search).parameters) - {"self"} + assert params == {"source_id", "topic", "limit"}, params + + +def test_requests_carry_no_api_keys_tokens_or_credentials(monkeypatch): + for name in ("OPENVERSE_API_KEY", "WIKIMEDIA_TOKEN", "API_KEY", "CLIENT_SECRET"): + monkeypatch.setenv(name, "should-never-be-sent") + + transport = FakeTransport([json_response(COMMONS_PAYLOAD)]) + registry = build_registry(client=make_client(transport)) + for source_id in registry.available_sources(): + transport_calls = len(transport.calls) + try: + registry.search(source_id, "cats", limit=1) + except SourceAccessError: # payload shapes differ per source; URL is what matters + pass + new_calls = transport.calls[transport_calls:] + assert new_calls, source_id + for call in new_calls: + blob = call["url"] + " " + repr(call["headers"]) + for needle in ("should-never-be-sent", "api_key", "apikey", "token", "Authorization"): + assert needle.lower() not in blob.lower(), (source_id, needle, blob) + + +def test_module_reads_no_credentials_from_environment(monkeypatch): + for name in ("OPENVERSE_API_KEY", "WIKIMEDIA_TOKEN", "API_KEY", "CLIENT_SECRET"): + monkeypatch.delattr("os.environ", name, raising=False) + first = build_registry() + assert first.available_sources() == ["openverse", "wikimedia_commons"] + monkeypatch.setenv("OPENVERSE_API_KEY", "secret") + second = build_registry() + assert second.available_sources() == first.available_sources() diff --git a/tools/errors.py b/tools/errors.py new file mode 100644 index 0000000..013ae8b --- /dev/null +++ b/tools/errors.py @@ -0,0 +1,187 @@ +"""Typed errors raised deliberately by the toolkit. + +The image-source layer never degrades a failed or unknown source into a silent, +empty result. Every such situation raises a :class:`ToolkitError` subclass that +carries a stable ``code`` string plus machine-readable ``details`` so the CLI can +map it onto an exit status and a structured JSON envelope. + +Hierarchy:: + + ToolkitError + +-- SourceError + +-- UnknownSourceError (unknown source id, lists available ids) + +-- DuplicateSourceError (registry already holds that id) + +-- SearchRequestError (topic/limit rejected before any request) + +-- SourceAccessError (the source could not be used) + +-- SourceUnreachableError (transport/DNS/timeout failure) + +-- SourceHTTPError (non-retryable or exhausted HTTP status) + +-- SourcePayloadError (body is not the expected JSON shape) + +-- HostNotAllowedError (URL outside the approved hosts) +""" + +from __future__ import annotations + +from typing import Any, Iterable, Mapping + +__all__ = [ + "DuplicateSourceError", + "HostNotAllowedError", + "SearchRequestError", + "SourceAccessError", + "SourceError", + "SourceHTTPError", + "SourcePayloadError", + "SourceUnreachableError", + "ToolkitError", + "UnknownSourceError", +] + + +class ToolkitError(Exception): + """Base class for errors the toolkit raises on purpose.""" + + code = "toolkit_error" + + def __init__( + self, + message: str, + *, + code: str | None = None, + details: Mapping[str, Any] | None = None, + ) -> None: + super().__init__(message) + self.message = message + if code is not None: + self.code = code + self.details: dict[str, Any] = dict(details or {}) + + def to_dict(self) -> dict[str, Any]: + """Return the JSON-safe envelope fragment describing this error.""" + + payload: dict[str, Any] = {"code": self.code, "message": self.message} + if self.details: + payload["details"] = dict(self.details) + return payload + + def __str__(self) -> str: # pragma: no cover - trivial + return self.message + + +class SourceError(ToolkitError): + """Base class for every image-source problem.""" + + code = "source_error" + + +class UnknownSourceError(SourceError): + """Raised when the caller asks for a source id the registry does not know.""" + + code = "source_unknown" + + def __init__(self, source_id: str, available: Iterable[str]) -> None: + self.source_id = str(source_id) + self.available: tuple[str, ...] = tuple(sorted(str(item) for item in available)) + listing = ", ".join(self.available) if self.available else "" + super().__init__( + f"unknown image source {self.source_id!r}; available sources: {listing}", + details={"source_id": self.source_id, "available_sources": list(self.available)}, + ) + + +class DuplicateSourceError(SourceError): + """Raised when an adapter id is registered twice.""" + + code = "source_duplicate" + + def __init__(self, source_id: str) -> None: + self.source_id = str(source_id) + super().__init__( + f"image source {self.source_id!r} is already registered", + details={"source_id": self.source_id}, + ) + + +class SearchRequestError(SourceError): + """Raised when the caller supplies a topic/limit a source cannot honour.""" + + code = "search_invalid_request" + + def __init__(self, message: str, *, details: Mapping[str, Any] | None = None) -> None: + super().__init__(message, details=details) + + +class SourceAccessError(SourceError): + """Raised when a known source cannot be reached or refuses to answer.""" + + code = "source_access" + + def __init__( + self, + message: str, + *, + url: str | None = None, + source_id: str | None = None, + details: Mapping[str, Any] | None = None, + ) -> None: + merged: dict[str, Any] = dict(details or {}) + if url is not None: + merged["url"] = url + if source_id is not None: + merged["source_id"] = source_id + super().__init__(message, details=merged) + self.url = url + self.source_id = source_id + + +class SourceUnreachableError(SourceAccessError): + """Raised when the transport cannot deliver a request to the host at all.""" + + code = "source_unreachable" + + def __init__(self, url: str, *, source_id: str | None = None, reason: str | None = None) -> None: + message = f"image source host is unreachable: {url}" + if reason: + message = f"{message} ({reason})" + super().__init__(message, url=url, source_id=source_id, details={"reason": reason} if reason else None) + + +class SourceHTTPError(SourceAccessError): + """Raised for an HTTP status the client cannot recover from.""" + + code = "source_http_error" + + def __init__(self, url: str, status: int, *, source_id: str | None = None) -> None: + super().__init__( + f"image source returned HTTP status {status}: {url}", + url=url, + source_id=source_id, + details={"status": status}, + ) + self.status = int(status) + + +class SourcePayloadError(SourceAccessError): + """Raised when a source answers with something the adapter cannot read.""" + + code = "source_invalid_response" + + def __init__(self, url: str, *, source_id: str | None = None, reason: str | None = None) -> None: + message = f"image source returned an unusable response: {url}" + if reason: + message = f"{message} ({reason})" + super().__init__(message, url=url, source_id=source_id, details={"reason": reason} if reason else None) + + +class HostNotAllowedError(SourceAccessError): + """Raised when a URL leaves the approved HTTPS host allowlist.""" + + code = "source_host_not_allowed" + + def __init__(self, url: str, *, allowed: Iterable[str] = ()) -> None: + self.allowed: tuple[str, ...] = tuple(sorted(str(item) for item in allowed)) + super().__init__( + f"refusing to contact {url!r}: only approved HTTPS hosts are allowed " + f"({', '.join(self.allowed) if self.allowed else 'none configured'})", + url=str(url), + details={"allowed_hosts": list(self.allowed)}, + ) diff --git a/tools/sources.py b/tools/sources.py new file mode 100644 index 0000000..5aa469a --- /dev/null +++ b/tools/sources.py @@ -0,0 +1,578 @@ +"""Pluggable, credential-free image source registry and its HTTPS layer. + +Nothing in this module reads an API key, token, registration or secret from +arguments, environment variables or configuration files: Wikimedia Commons and +Openverse are queried anonymously. The module provides + +* :class:`HTTPSClient` -- a small HTTPS client that identifies itself with a + stable ``User-Agent``, refuses every URL outside the approved hosts + (Wikimedia Commons and Openverse) on port 443, retries HTTP 429/5xx a bounded + number of times with capped exponential backoff, and raises + :class:`~tools.errors.SourceAccessError` instead of returning an empty result + when the host is unreachable or keeps failing; +* :class:`SourceAdapter` -- the interface every image source implements; +* :class:`WikimediaCommonsAdapter` and :class:`OpenverseAdapter` -- two working + credential-free sources behind that interface; +* :class:`SourceRegistry` -- the plugin list that reports available source ids + and turns an unknown id into :class:`~tools.errors.UnknownSourceError` listing + them. +""" + +from __future__ import annotations + +import json +import time +import urllib.error +import urllib.parse +import urllib.request +from dataclasses import dataclass, field +from typing import Any, Callable, Iterable, Mapping, Protocol, runtime_checkable + +from .errors import ( + DuplicateSourceError, + HostNotAllowedError, + SearchRequestError, + SourceAccessError, + SourceHTTPError, + SourceError, + SourcePayloadError, + SourceUnreachableError, + UnknownSourceError, +) + +__all__ = [ + "APPROVED_HOSTS", + "DEFAULT_LIMIT", + "DEFAULT_MAX_ATTEMPTS", + "HTTPSClient", + "HTTPResponse", + "MAX_LIMIT", + "OPENVERSE_API_HOST", + "OpenverseAdapter", + "SourceAdapter", + "SourceCandidate", + "SourceRegistry", + "USER_AGENT", + "WIKIMEDIA_COMMONS_API_HOST", + "WIKIMEDIA_COMMONS_MEDIA_HOST", + "build_registry", + "default_registry", + "is_approved_url", + "search", + "validate_source_url", +] + +PRODUCT_NAME = "product-54354138-1" +USER_AGENT = f"{PRODUCT_NAME}-image-toolkit/0.1.0 (credential-free image source client)" + +WIKIMEDIA_COMMONS_API_HOST = "commons.wikimedia.org" +WIKIMEDIA_COMMONS_MEDIA_HOST = "upload.wikimedia.org" +OPENVERSE_API_HOST = "api.openverse.org" + +# Wikimedia Commons is reachable through its API host plus the media host that +# serves the original files, Openverse through its API host (and the thumbnail +# endpoint on the same API host). +APPROVED_HOSTS = frozenset( + {WIKIMEDIA_COMMONS_API_HOST, WIKIMEDIA_COMMONS_MEDIA_HOST, OPENVERSE_API_HOST} +) +APPROVED_SCHEME = "https" +APPROVED_PORT = 443 + +DEFAULT_TIMEOUT = 10.0 +DEFAULT_MAX_ATTEMPTS = 3 +DEFAULT_BACKOFF_SECONDS = 0.5 +MAX_BACKOFF_SECONDS = 8.0 +RETRY_STATUS_CODES = frozenset({429}) | frozenset(range(500, 600)) + + +@dataclass(frozen=True) +class HTTPResponse: + """A transport-independent HTTP response.""" + + status: int + body: bytes + headers: Mapping[str, str] = field(default_factory=dict) + + def json(self, *, url: str = "", source_id: str | None = None) -> Any: + try: + return json.loads(self.body) + except (ValueError, UnicodeDecodeError) as exc: # pragma: no cover - defensive + raise SourcePayloadError(url or "", source_id=source_id, reason=str(exc)) from exc + + +class Transport(Protocol): + """Callable performing one HTTP GET; injectable so tests never hit the net.""" + + def __call__(self, url: str, headers: Mapping[str, str], timeout: float) -> HTTPResponse: ... + + +def validate_source_url(url: str, *, allowed: Iterable[str] = APPROVED_HOSTS) -> str: + """Return *url* when it is an approved HTTPS URL, else raise a typed error.""" + + allowed_set = frozenset(allowed) + if not isinstance(url, str) or not url: + raise HostNotAllowedError(str(url), allowed=allowed_set) + parts = urllib.parse.urlsplit(url) + if parts.scheme != APPROVED_SCHEME: + raise HostNotAllowedError(url, allowed=allowed_set) + host = (parts.hostname or "").lower() + if host not in allowed_set: + raise HostNotAllowedError(url, allowed=allowed_set) + try: + port = parts.port + except ValueError as exc: + raise HostNotAllowedError(url, allowed=allowed_set) from exc + if port not in (None, APPROVED_PORT): + raise HostNotAllowedError(url, allowed=allowed_set) + return url + + +def is_approved_url(url: str, *, allowed: Iterable[str] = APPROVED_HOSTS) -> bool: + """Non-raising variant of :func:`validate_source_url` used by tests/callers.""" + + try: + validate_source_url(url, allowed=allowed) + except HostNotAllowedError: + return False + return True + + +def urllib_transport(url: str, headers: Mapping[str, str], timeout: float) -> HTTPResponse: + """Default transport: a plain stdlib GET returning status, body and headers.""" + + request = urllib.request.Request(url, headers=dict(headers), method="GET") + try: + with urllib.request.urlopen(request, timeout=timeout) as response: + return HTTPResponse( + status=int(getattr(response, "status", 200)), + body=response.read(), + headers=dict(response.headers.items()), + ) + except urllib.error.HTTPError as exc: + body = b"" + try: + body = exc.read() + except Exception: # pragma: no cover - body optional + body = b"" + return HTTPResponse( + status=int(exc.code), + body=body, + headers=dict(getattr(exc, "headers", {}) or {}), + ) + + +class HTTPSClient: + """Fetch a URL only when it is approved, with bounded retry/backoff. + + The client never returns an empty result for a broken source: unreachable + hosts raise :class:`SourceUnreachableError` and exhausted/non-retryable + statuses raise :class:`SourceHTTPError`. Both derive from + :class:`SourceAccessError`. + """ + + def __init__( + self, + *, + user_agent: str = USER_AGENT, + allowed_hosts: Iterable[str] = APPROVED_HOSTS, + timeout: float = DEFAULT_TIMEOUT, + max_attempts: int = DEFAULT_MAX_ATTEMPTS, + backoff_seconds: float = DEFAULT_BACKOFF_SECONDS, + max_backoff_seconds: float = MAX_BACKOFF_SECONDS, + transport: Transport | None = None, + sleep: Callable[[float], None] = time.sleep, + ) -> None: + if max_attempts < 1: + raise ValueError("max_attempts must be >= 1") + self.user_agent = user_agent + self.allowed_hosts = frozenset(allowed_hosts) + self.timeout = float(timeout) + self.max_attempts = int(max_attempts) + self.backoff_seconds = float(backoff_seconds) + self.max_backoff_seconds = float(max_backoff_seconds) + self._transport: Transport = transport or urllib_transport + self._sleep = sleep + + # -- public API ----------------------------------------------------- + def headers(self) -> dict[str, str]: + """Request headers, always including the identifying User-Agent.""" + + return { + "User-Agent": self.user_agent, + "Accept": "application/json, image/*;q=0.8, */*;q=0.5", + } + + def backoff_delay(self, attempt: int) -> float: + """Capped exponential backoff delay for 1-based *attempt*.""" + + delay = self.backoff_seconds * (2 ** max(0, attempt - 1)) + return min(delay, self.max_backoff_seconds) + + def get(self, url: str, *, source_id: str | None = None) -> HTTPResponse: + """GET *url*, retrying transient statuses, then raise typed errors.""" + + validate_source_url(url, allowed=self.allowed_hosts) + response: HTTPResponse | None = None + for attempt in range(1, self.max_attempts + 1): + try: + response = self._transport(url, self.headers(), self.timeout) + except SourceAccessError: + raise + except Exception as exc: # transport-level failure + if attempt >= self.max_attempts: + raise SourceUnreachableError( + url, source_id=source_id, reason=f"{type(exc).__name__}: {exc}" + ) from exc + self._sleep(self.backoff_delay(attempt)) + continue + if response.status in RETRY_STATUS_CODES and attempt < self.max_attempts: + self._sleep(self.backoff_delay(attempt)) + continue + if 200 <= response.status < 300: + return response + raise SourceHTTPError(url, response.status, source_id=source_id) + # Defensive: the loop either returned or raised above. + status = response.status if response is not None else 0 + raise SourceHTTPError(url, status, source_id=source_id) + + def get_json(self, url: str, *, source_id: str | None = None) -> Any: + response = self.get(url, source_id=source_id) + try: + return json.loads(response.body) + except (ValueError, UnicodeDecodeError) as exc: + raise SourcePayloadError(url, source_id=source_id, reason=str(exc)) from exc + + +DEFAULT_LIMIT = 10 +MAX_LIMIT = 50 + + +@dataclass(frozen=True) +class SourceCandidate: + """One normalized search hit. + + ``source_id`` identifies the adapter, ``file_url`` is the direct HTTPS + location of the original image, ``mime_type`` is the type the source + reports (may be ``None`` when unstated) and ``fetch_bytes`` is a zero-argument + callable returning the image bytes through the approved HTTPS client. + """ + + source_id: str + file_url: str + mime_type: str | None + fetch_bytes: Callable[[], bytes] + title: str | None = None + page_url: str | None = None + width: int | None = None + height: int | None = None + extra: Mapping[str, Any] = field(default_factory=dict) + + def fetch(self) -> bytes: + """Convenience passthrough to :attr:`fetch_bytes`.""" + + return self.fetch_bytes() + + def to_dict(self) -> dict[str, Any]: + """Metadata-only projection (the callable is intentionally omitted).""" + + payload: dict[str, Any] = { + "source_id": self.source_id, + "file_url": self.file_url, + "mime_type": self.mime_type, + } + for key in ("title", "page_url", "width", "height"): + value = getattr(self, key) + if value is not None: + payload[key] = value + return payload + + +@runtime_checkable +class SourceAdapter(Protocol): + """Interface implemented by every image source.""" + + source_id: str + + def search(self, topic: str, limit: int) -> list[SourceCandidate]: ... + + +def normalize_search_args(topic: str, limit: int) -> tuple[str, int]: + """Validate the caller-supplied topic and result limit.""" + + if not isinstance(topic, str): + raise SearchRequestError( + f"topic must be a string, got {type(topic).__name__}", + details={"topic": repr(topic)}, + ) + topic = topic.strip() + if not topic: + raise SearchRequestError("topic must not be empty", details={"topic": topic}) + if isinstance(limit, bool) or not isinstance(limit, int): + raise SearchRequestError( + f"limit must be an integer, got {type(limit).__name__}", + details={"limit": repr(limit)}, + ) + if limit < 1: + raise SearchRequestError(f"limit must be >= 1, got {limit}", details={"limit": limit}) + if limit > MAX_LIMIT: + raise SearchRequestError( + f"limit must be <= {MAX_LIMIT}, got {limit}", details={"limit": limit} + ) + return topic, limit + + +class _BaseAdapter: + """Shared normalization helpers for concrete adapters.""" + + source_id = "" + + def __init__(self, client: HTTPSClient | None = None) -> None: + self.client = client or HTTPSClient() + + def _candidate( + self, + *, + file_url: str, + mime_type: str | None, + title: str | None = None, + page_url: str | None = None, + width: int | None = None, + height: int | None = None, + extra: Mapping[str, Any] | None = None, + validate: bool = True, + ) -> SourceCandidate: + if validate: + validate_source_url(file_url, allowed=self.client.allowed_hosts) + source_id = self.source_id + client = self.client + return SourceCandidate( + source_id=source_id, + file_url=file_url, + mime_type=mime_type, + fetch_bytes=lambda url=file_url: client.get(url, source_id=source_id).body, + title=title, + page_url=page_url, + width=width, + height=height, + extra=dict(extra or {}), + ) + + def __repr__(self) -> str: # pragma: no cover - trivial + return f"<{type(self).__name__} source_id={self.source_id!r}>" + + +class WikimediaCommonsAdapter(_BaseAdapter): + """Wikimedia Commons search through the anonymous MediaWiki action API.""" + + source_id = "wikimedia_commons" + api_host = WIKIMEDIA_COMMONS_API_HOST + + def build_search_url(self, topic: str, limit: int) -> str: + """Construct the generator search URL for *topic* and *limit*.""" + + topic, limit = normalize_search_args(topic, limit) + params = { + "action": "query", + "format": "json", + "formatversion": "2", + "generator": "search", + "gsrsearch": topic, + "gsrnamespace": "6", + "gsrlimit": str(limit), + "prop": "imageinfo", + "iiprop": "url|mime|size", + "iiurlwidth": "1024", + } + query = urllib.parse.urlencode(params) + return f"https://{self.api_host}/w/api.php?{query}" + + def search(self, topic: str, limit: int) -> list[SourceCandidate]: + url = self.build_search_url(topic, limit) + payload = self.client.get_json(url, source_id=self.source_id) + if not isinstance(payload, Mapping): + raise SourcePayloadError(url, source_id=self.source_id, reason="expected a JSON object") + if "error" in payload: + raise SourceAccessError( + f"Wikimedia Commons API error: {payload['error']}", + url=url, + source_id=self.source_id, + ) + pages = (payload.get("query") or {}).get("pages") or [] + candidates: list[SourceCandidate] = [] + for page in pages: + infos = page.get("imageinfo") or [] + if not infos: + continue + info = infos[0] + file_url = info.get("url") + if not file_url: + continue + candidates.append( + self._candidate( + file_url=file_url, + mime_type=info.get("mime"), + title=page.get("title"), + page_url=info.get("descriptionurl"), + width=info.get("width"), + height=info.get("height"), + extra={"thumb_url": info.get("thumburl")} if info.get("thumburl") else None, + ) + ) + if len(candidates) >= limit: + break + return candidates[:limit] + + +class OpenverseAdapter(_BaseAdapter): + """Openverse search through its anonymous public images endpoint.""" + + source_id = "openverse" + api_host = OPENVERSE_API_HOST + + def build_search_url(self, topic: str, limit: int) -> str: + """Construct the Openverse image search URL for *topic* and *limit*.""" + + topic, limit = normalize_search_args(topic, limit) + query = urllib.parse.urlencode({"q": topic, "page_size": str(limit)}) + return f"https://{self.api_host}/v1/images/?{query}" + + def search(self, topic: str, limit: int) -> list[SourceCandidate]: + url = self.build_search_url(topic, limit) + payload = self.client.get_json(url, source_id=self.source_id) + if not isinstance(payload, Mapping): + raise SourcePayloadError(url, source_id=self.source_id, reason="expected a JSON object") + results = payload.get("results") + if results is None: + raise SourcePayloadError(url, source_id=self.source_id, reason="missing 'results'") + candidates: list[SourceCandidate] = [] + for item in results: + if not isinstance(item, Mapping): + continue + original_url = item.get("url") + file_url = item.get("thumbnail") or original_url + if not file_url: + continue + candidates.append( + self._candidate( + file_url=file_url, + # Origin files may live on third-party hosts; the client + # still refuses them at fetch time with a typed error. + validate=is_approved_url(file_url, allowed=self.client.allowed_hosts), + mime_type=item.get("filetype"), + title=item.get("title"), + page_url=item.get("foreign_landing_url") or item.get("detail_url"), + width=item.get("width"), + height=item.get("height"), + extra={ + "license": item.get("license"), + "creator": item.get("creator"), + "original_url": original_url, + }, + ) + ) + if len(candidates) >= limit: + break + return candidates[:limit] + + +class SourceRegistry: + """Ordered plugin registry of credential-free image sources.""" + + def __init__(self, adapters: Iterable[SourceAdapter] | None = None) -> None: + self._adapters: dict[str, SourceAdapter] = {} + for adapter in adapters or (): + self.register(adapter) + + # -- registration --------------------------------------------------- + def register(self, adapter: SourceAdapter, *, replace: bool = False) -> SourceAdapter: + """Add *adapter*; a duplicate id raises :class:`DuplicateSourceError`.""" + + source_id = getattr(adapter, "source_id", None) + if not isinstance(source_id, str) or not source_id: + raise SourceError("adapter must expose a non-empty string 'source_id'") + if source_id in self._adapters and not replace: + raise DuplicateSourceError(source_id) + self._adapters[source_id] = adapter + return adapter + + def unregister(self, source_id: str) -> SourceAdapter: + if source_id not in self._adapters: + raise UnknownSourceError(source_id, self.available_sources()) + return self._adapters.pop(source_id) + + # -- lookup --------------------------------------------------------- + def available_sources(self) -> list[str]: + """Sorted ids of every registered source.""" + + return sorted(self._adapters) + + def __contains__(self, source_id: object) -> bool: + return isinstance(source_id, str) and source_id in self._adapters + + def __len__(self) -> int: + return len(self._adapters) + + def __iter__(self): + return iter(self.available_sources()) + + def get(self, source_id: str) -> SourceAdapter: + """Return the adapter for *source_id* or raise a listing error.""" + + try: + return self._adapters[source_id] + except (KeyError, TypeError): + raise UnknownSourceError(source_id, self.available_sources()) from None + + def create(self, source_id: str) -> SourceAdapter: + """Explicit alias of :meth:`get` used by the CLI layer.""" + + return self.get(source_id) + + # -- search --------------------------------------------------------- + def search(self, source_id: str, topic: str, limit: int = DEFAULT_LIMIT) -> list[SourceCandidate]: + """Search *source_id* for *topic*, returning at most *limit* candidates.""" + + adapter = self.get(source_id) + topic, limit = normalize_search_args(topic, limit) + return list(adapter.search(topic, limit)) + + def __repr__(self) -> str: # pragma: no cover - trivial + return f"SourceRegistry({self.available_sources()!r})" + + + + +def build_registry( + *, + client: HTTPSClient | None = None, + adapters: Iterable[SourceAdapter] | None = None, +) -> SourceRegistry: + """Build the default registry (Wikimedia Commons + Openverse).""" + + shared_client = client or HTTPSClient() + if adapters is None: + adapters = ( + WikimediaCommonsAdapter(shared_client), + OpenverseAdapter(shared_client), + ) + return SourceRegistry(adapters) + + +def default_registry(client: HTTPSClient | None = None) -> SourceRegistry: + """Convenience wrapper returning a fresh default registry.""" + + return build_registry(client=client) + + +def search( + topic: str, + *, + source: str = "wikimedia_commons", + limit: int = DEFAULT_LIMIT, + registry: SourceRegistry | None = None, + client: HTTPSClient | None = None, +) -> list[SourceCandidate]: + """Search *source* for *topic* using *registry* (default registry by default).""" + + active = registry if registry is not None else build_registry(client=client) + return active.search(source, topic, limit)