devplacepy/controller.py at e6698b11f928dda7586ca52dc70cda27198e6b7b

 # retoor <retoor@molodetz.nl>
 from __future__ import annotations
 import asyncio
 import ipaddress
 import json
 import logging
 import re
 import socket
 from typing import Any
 from urllib.parse import urlparse
 import httpx
 from ..config import Settings
 from ..errors import NetworkError, ToolInputError, UpstreamError
 from ..text import html_to_text
 logger = logging.getLogger("devii.fetch")
 CHROME_VERSION = "131"
 CHROME_VERSION_FULL = "131.0.0.0"
 STEALTH_HEADERS = {
     "User-Agent": (
         "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
         f"(KHTML, like Gecko) Chrome/{CHROME_VERSION_FULL} Safari/537.36"
     ),
     "Accept": (
         "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,"
         "image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7"
     ),
     "Accept-Language": "en-US,en;q=0.9",
     "sec-ch-ua": (
         f'"Google Chrome";v="{CHROME_VERSION}", "Chromium";v="{CHROME_VERSION}", '
         '"Not_A Brand";v="24"'
     ),
     "sec-ch-ua-mobile": "?0",
     "sec-ch-ua-platform": '"Linux"',
     "Sec-Fetch-Dest": "document",
     "Sec-Fetch-Mode": "navigate",
     "Sec-Fetch-Site": "none",
     "Sec-Fetch-User": "?1",
     "Upgrade-Insecure-Requests": "1",
     "Cache-Control": "max-age=0",
     "DNT": "1",
 }
 TITLE = re.compile(r"<title[^>]*>(.*?)</title>", re.IGNORECASE | re.DOTALL)
 MIN_FETCH_CHARS = 1000
 NAT64_PREFIXES = (
     ipaddress.ip_network("64:ff9b::/96"),
     ipaddress.ip_network("64:ff9b:1::/48"),
 )
 def _effective_address(address: Any) -> Any:
     """Resolve NAT64 / IPv4-mapped IPv6 to the embedded IPv4 so DNS64 networks
     don't make public sites look like reserved addresses."""
     if isinstance(address, ipaddress.IPv6Address):
         if address.ipv4_mapped is not None:
             return address.ipv4_mapped
         for prefix in NAT64_PREFIXES:
             if address in prefix:
                 return ipaddress.IPv4Address(int(address) & 0xFFFFFFFF)
     return address
 class FetchController:
     def __init__(self, settings: Settings) -> None:
         self._settings = settings
     async def dispatch(self, name: str, arguments: dict[str, Any]) -> str:
         if name != "fetch_url":
             raise ToolInputError(f"Unknown fetch tool: {name}")
         return await self._fetch_url(arguments)
     def _optional_cap(self, requested: Any, content: str) -> str:
         if requested is None:
             return content
         try:
             cap = max(MIN_FETCH_CHARS, int(requested))
         except (TypeError, ValueError):
             return content
         return content[:cap]
     async def _fetch_url(self, arguments: dict[str, Any]) -> str:
         url = str(arguments.get("url", "")).strip()
         if not url:
             raise ToolInputError("fetch_url requires a url.")
         if "://" not in url:
             url = "https://" + url
         await self._guard(url)
         body, final_url, content_type, status = await self._download(url)
         if "html" in content_type or "xml" in content_type or content_type.startswith("text/"):
             title_match = TITLE.search(body)
             title = html_to_text(title_match.group(1)) if title_match else ""
             content = html_to_text(body)
         elif "json" in content_type:
             title = ""
             content = body
         else:
             title = ""
             content = f"(non-text response: {content_type})"
         return json.dumps(
             {
                 "status": "success",
                 "url": final_url,
                 "http_status": status,
                 "content_type": content_type,
                 "title": title,
                 "content": self._optional_cap(arguments.get("max_chars"), content),
             },
             ensure_ascii=False,
         )
     async def _guard(self, url: str) -> None:
         parsed = urlparse(url)
         if parsed.scheme not in ("http", "https"):
             raise ToolInputError("Only http and https URLs can be fetched.")
         host = parsed.hostname
         if not host:
             raise ToolInputError("URL has no host.")
         if self._settings.fetch_allow_private:
             return
         try:
             infos = await asyncio.to_thread(socket.getaddrinfo, host, None)
         except socket.gaierror as exc:
             raise NetworkError(f"Could not resolve host: {host}", host=host) from exc
         for info in infos:
             address = _effective_address(ipaddress.ip_address(info[4][0]))
             if (
                 address.is_private
                 or address.is_loopback
                 or address.is_link_local
                 or address.is_reserved
                 or address.is_multicast
                 or address.is_unspecified
             ):
                 raise ToolInputError(
                     f"Refusing to fetch a private or local address ({address}). "
                     "Set DEVII_FETCH_ALLOW_PRIVATE=1 to override."
                 )
     async def _download(self, url: str) -> tuple[str, str, str, int]:
         limit = self._settings.fetch_max_bytes
         try:
             async with httpx.AsyncClient(
                 headers=STEALTH_HEADERS,
                 follow_redirects=True,
                 timeout=self._settings.fetch_timeout_seconds,
             ) as client:
                 async with client.stream("GET", url) as response:
                     if response.status_code >= 500:
                         raise UpstreamError(
                             f"Server returned {response.status_code}.",
                             status=response.status_code,
                             url=url,
                         )
                     chunks: list[bytes] = []
                     total = 0
                     async for chunk in response.aiter_bytes():
                         chunks.append(chunk)
                         total += len(chunk)
                         if total >= limit:
                             break
                     raw = b"".join(chunks)[:limit]
                     encoding = response.encoding or "utf-8"
                     try:
                         body = raw.decode(encoding, errors="replace")
                     except LookupError:
                         body = raw.decode("utf-8", errors="replace")
                     content_type = response.headers.get("content-type", "").lower()
                     return body, str(response.url), content_type, response.status_code
         except httpx.TimeoutException as exc:
             raise NetworkError(f"Request timed out fetching {url}", url=url) from exc
         except httpx.HTTPError as exc:
             raise NetworkError(f"Could not fetch {url}: {exc}", url=url) from exc