signal-bot/signal_bot/http_client.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Outbound HTTP for the image commands.

Every outbound request goes through this one injectable client, so the tests
and the offline environment can serve fixtures without the bot knowing.
"""

from __future__ import annotations

import base64
from typing import TYPE_CHECKING, Final

if TYPE_CHECKING:
    from collections.abc import Sequence

    import httpx


class ImageFetchError(RuntimeError):
    """Raised when an upstream image API did not yield a usable image."""


class ImageFetcher:
    """Resolves an image URL from a JSON API and returns it base64-encoded."""

    def __init__(self, client: httpx.AsyncClient) -> None:
        """Store the injected HTTP client."""
        self._client = client

    async def fetch_base64(
        self,
        api_url: str,
        url_path: Sequence[str | int],
        rewrite: tuple[str, str] | None = None,
    ) -> str:
        """Fetch an image and return it base64-encoded, ready for Signal."""
        return base64.b64encode(await self.fetch_bytes(api_url, url_path, rewrite)).decode("ascii")

    async def fetch_bytes(
        self,
        api_url: str,
        url_path: Sequence[str | int],
        rewrite: tuple[str, str] | None = None,
        params: dict[str, str] | None = None,
    ) -> bytes:
        """Fetch an image and return its raw bytes.

        Args:
            api_url: JSON endpoint that names an image.
            url_path: Keys/indices to walk in the JSON to reach the image URL.
            rewrite: Optional (find, replace) pair applied to the URL found
                there. Some APIs name a thumbnail and leave the full-size
                photo one substitution away -- iNaturalist's observation JSON
                gives a 75px `/square.` URL whose `/large.` sibling is the
                real photo -- and a substitution keeps that upstream quirk as
                data next to the endpoint rather than as a branch in here.

            params: Query string for the JSON endpoint, when it is easier to
                pass than to embed -- the Commons API takes a title that has
                to be percent-encoded, the animal APIs are fixed URLs.

        Returns:
            The image bytes.

        Raises:
            ImageFetchError: The JSON did not contain a usable URL.
        """
        response = await self._client.get(api_url, params=params)
        response.raise_for_status()
        image_url = _walk(response.json(), url_path)
        if rewrite is not None:
            image_url = image_url.replace(*rewrite)

        # These bytes are never written to a path built from the remote URL:
        # the original bot named a local file after the URL's last path
        # segment and wrote it into the process CWD, which is both path
        # traversal and unbounded. portraits.py does cache to disk, but under
        # a name it derives from the pope's ordinal, never from upstream.
        image = await self._client.get(image_url)
        image.raise_for_status()
        return image.content


_MISSING: Final = object()


def _in_range(items: list[object], index: int) -> bool:
    """True when index addresses a real element, counting from either end."""
    return -len(items) <= index < len(items)


def _step(node: object, key: str | int, path: Sequence[str | int]) -> object:
    """Descend one level into a JSON document.

    Narrows explicitly instead of indexing blindly, so a malformed upstream
    response raises ImageFetchError rather than a TypeError from deeper in.
    """
    found: object = _MISSING
    if isinstance(key, str) and isinstance(node, dict):
        found = node.get(key, _MISSING)
    elif isinstance(key, int) and isinstance(node, list) and _in_range(node, key):
        found = node[key]
    if found is _MISSING:
        msg = f"image API response has no element at {path!r}"
        raise ImageFetchError(msg)
    return found


def _walk(document: object, path: Sequence[str | int]) -> str:
    """Walk a JSON document along path and return the string found there."""
    node: object = document
    for key in path:
        node = _step(node, key, path)
    if not isinstance(node, str) or not node:
        msg = f"image API element at {path!r} is not a URL: {node!r}"
        raise ImageFetchError(msg)
    return node