app/services/fetcher.py

"""Safe async page fetching for the asset extractor.

Guards: SSRF block-list, redirect cap, size cap, timeout, content-type check.
"""

from urllib.parse import urlparse

import httpx

from app.config import (
    ASSETS_BLOCKED_HOSTS,
    ASSETS_MAX_BYTES,
    ASSETS_MAX_REDIRECTS,
    ASSETS_FETCH_TIMEOUT,
    ASSETS_USER_AGENT,
)


class FetchError(Exception):
    """Raised when the page cannot be safely fetched."""


def _blocked(raw_url: str) -> str | None:
    """Return a reason string if the URL host is disallowed, else None."""
    parsed = urlparse(raw_url)
    host = (parsed.hostname or "").lower()
    if host in ASSETS_BLOCKED_HOSTS:
        return "blocked private/local host"
    if parsed.scheme not in ("http", "https"):
        return f"unsupported scheme: {parsed.scheme}"
    return None


async def fetch_html(url: str) -> tuple[str, str, str]:
    """Fetch a page and return (final_url, html, title).

    Raises FetchError on any unsafe/failed/oversized/non-HTML fetch.
    """

    # Basic SSRF / scheme guard on the requested URL.
    reason = _blocked(url)
    if reason:
        raise FetchError(reason)

    headers = {
        "User-Agent": ASSETS_USER_AGENT,
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    }

    limits = httpx.Limits(max_connections=50, max_keepalive_connections=10)

    try:
        async with httpx.AsyncClient(
            follow_redirects=True,
            max_redirects=ASSETS_MAX_REDIRECTS,
            timeout=ASSETS_FETCH_TIMEOUT,
            limits=limits,
            headers=headers,
        ) as client:
            async with client.stream("GET", url) as resp:
                resp.raise_for_status()

                content_type = resp.headers.get("content-type", "").lower()
                if "html" not in content_type and "xml" not in content_type:
                    raise FetchError(f"not an HTML page (content-type: {content_type or 'unknown'})")

                # Enforce size cap while streaming.
                chunks: list[bytes] = []
                size = 0
                async for chunk in resp.aiter_bytes():
                    size += len(chunk)
                    if size > ASSETS_MAX_BYTES:
                        raise FetchError(f"page exceeds {ASSETS_MAX_BYTES} bytes")
                    chunks.append(chunk)

                final_url = str(resp.url)
                html = b"".join(chunks).decode("utf-8", "ignore")
    except httpx.HTTPStatusError as exc:
        raise FetchError(f"HTTP error {exc.response.status_code}") from exc
    except httpx.HTTPError as exc:
        raise FetchError(f"request failed: {exc}") from exc

    # Guard against redirected-to-private hosts (defence in depth).
    reason = _blocked(final_url)
    if reason:
        raise FetchError(f"redirected to {reason}")

    title = _extract_title(html)
    return final_url, html, title


def _extract_title(html: str) -> str | None:
    import re

    m = re.search(r"<title[^>]*>(.*?)</title>", html, re.IGNORECASE | re.DOTALL)
    if m:
        return m.group(1).strip()
    return None