LangBot/plugins/misc/WebSearch/components/tools/sites/safe_fetch.py

"""Target-validated, byte-bounded HTTP(S) fetching for the site adapters.

Why this module exists
----------------------
``visit_web`` accepts an arbitrary user-supplied URL and runs inside the shared
runtime, where one process serves every tenant. A plain ``requests.get(url)``
had two blocking problems (certification review F1/F2):

* the caller picked the server-side network target - loopback, link-local
  metadata (``169.254.169.254``), private ranges, or a public host that
  redirects to one of them - and the fetch inherited the environment's proxy
  /netrc credentials;
* the whole response was buffered and parsed before the ``brief_len``
  truncation, so a hostile server could occupy shared memory and a shared
  worker thread without bound (``timeout`` only bounds a single socket wait).

This module therefore:

* accepts only ``http``/``https`` URLs;
* resolves the host itself, refuses every resolution that contains a
  non-public address, and then connects to the *validated IP literal* - a
  later, different DNS answer can no longer rebind the connection;
* follows redirects manually, re-running the same validation on every hop;
* reads the body in bounded chunks with a hard cap on the decoded bytes *and* a
  cumulative hard cap on the encoded (still compressed) bytes read, a total
  deadline and a cooperative stop flag, and closes the socket on every exit
  path;
* handles the decompression stream's end explicitly: exactly one gzip/deflate
  member is supported, bytes that follow the member raise
  :class:`TrailingDataError` instead of accumulating in zlib's ``unused_data``,
  an end-of-stream decompressor is never fed again, and a body whose compressed
  stream stops before its end marker raises
  :class:`IncompleteCompressedStreamError` (so the gzip CRC32 / zlib Adler-32 is
  validated before the body can become a document);
* never returns a partial body as if it were complete: every exit path re-checks
  the stop flag and the deadline, and a body shorter than its declared
  ``Content-Length`` fails with :class:`TruncatedResponseError`;
* enforces that deadline and stop flag on *every* blocking step - DNS, connect,
  TLS handshake, status/header parsing and body reads - by shutting the socket
  down from a watchdog thread, because a socket timeout only bounds the idle
  time between two reads and a peer that drips bytes can outlive it forever;
* runs every deadline-bounded ``getaddrinfo`` on a fixed resolver worker pool
  with a fixed in-flight budget, so a lookup that a waiter abandoned keeps its
  slot until it really ends and further lookups are refused with
  :class:`ResolverBusyError` instead of accumulating threads or queueing
  without limit;
* opens the socket directly, so no proxy environment variables, netrc or
  ``requests``-level authentication defaults are consulted;
* verifies TLS with the default CA store and the original hostname (the
  pinned IP is only the transport target);
* never puts a raw URL back into an error message: :func:`redact_url` drops the
  userinfo, query string and fragment, and the same scrub is applied to
  chained exception text so a traceback cannot leak credentials either.

Limits and deliberate divergence
--------------------------------
The encoded-byte budget (:data:`MAX_ENCODED_BYTES`) covers bytes *after* the
first compressed member, the trailing-data refusal and the incomplete-stream
check.  These belong to this plugin's ``safe_fetch.py`` only.  ``misc/URLSummary``
ships an independent copy of this module whose 0.1.9 artifact is already
certified; that copy keeps the latent trailing-data hole and is intentionally
*not* updated here, so the two files diverge on this point by design (flagged to
the maintainer).
"""

from __future__ import annotations

import base64
import collections
import dataclasses
import http.client
import ipaddress
import re
import socket
import ssl
import threading
import time
import zlib
from typing import Any, Iterable, Iterator, Mapping
from urllib.parse import unquote, urljoin, urlsplit, urlunsplit

MAX_RESPONSE_BYTES = 2 * 1024 * 1024
"""Hard cap on the decoded response body kept for one fetch."""

_ENCODED_SLACK = 64 * 1024
"""Framing slack added on top of a fetch's decoded cap to bound encoded bytes.

A well-formed gzip/DEFLATE stream that inflates to ``N`` bytes is essentially
``N`` bytes on the wire: DEFLATE stored blocks add ~5 bytes per 64 KiB payload
and a gzip wrapper adds a fixed ~18 bytes, so the encoding overhead is well
under 0.1%.  One whole read chunk (:data:`_CHUNK_SIZE`) of slack is therefore
far more than any legitimate stream needs, while the resulting cumulative
budget still stops a peer that keeps streaming encoded bytes after the first
compressed member has ended.
"""

MAX_ENCODED_BYTES = MAX_RESPONSE_BYTES + _ENCODED_SLACK
"""Default cumulative hard cap on the encoded bytes read for one fetch.

This is the value at the default 2 MiB decoded cap; a fetch with a smaller
``max_bytes`` gets a proportionally smaller encoded budget.
"""

MAX_PARSE_CHARS = 256 * 1024
"""Cap on the HTML text handed to BeautifulSoup."""

MAX_REDIRECTS = 5
CONNECT_TIMEOUT = 5.0
READ_TIMEOUT = 5.0
TOTAL_TIMEOUT = 15.0

MAX_DNS_WORKERS = 4
"""Fixed number of threads that run blocking ``getaddrinfo`` calls."""

MAX_DNS_OUTSTANDING = 8
"""Fixed cap on resolutions in flight, running and queued together."""

_CHUNK_SIZE = 64 * 1024
_REDIRECT_STATUSES = frozenset({301, 302, 303, 307, 308})
_DEFAULT_PORTS = {"http": 80, "https": 443}
_REQUEST_DEFAULTS = {
    "Accept": "text/html,application/xhtml+xml,*/*;q=0.8",
    "Accept-Encoding": "gzip, deflate",
    "Connection": "close",
}


class FetchError(Exception):
    """Base class for every failure raised by this module."""


class BlockedTargetError(FetchError):
    """The URL or one of its redirects points at a forbidden target."""


class ResponseTooLargeError(FetchError):
    """The response body exceeded the configured byte cap."""


class FetchAbortedError(FetchError):
    """The fetch was cancelled or ran past its deadline."""


class TooManyRedirectsError(FetchError):
    """The redirect chain is longer than :data:`MAX_REDIRECTS`."""


class TruncatedResponseError(FetchError):
    """The peer stopped before delivering the body it declared."""


class TrailingDataError(FetchError):
    """Unpermitted bytes followed the end of the compressed body's stream.

    The decoder supports exactly one gzip/DEFLATE member, so anything after
    the stream's end is refused rather than buffered in ``unused_data``.
    """


class IncompleteCompressedStreamError(FetchError):
    """A compressed body ended before its stream's end marker.

    The stream's integrity (gzip CRC32/length, zlib Adler-32) can only be
    checked at the end of a member, so a body that stops early must fail
    instead of being handed on as a document.
"""


class ResolverBusyError(FetchError):
    """The bounded resolver budget is exhausted; a lookup was refused."""


class _ByteCounter:
    """Counts raw (still encoded) body bytes and enforces a cumulative cap.

    The cap is checked before each chunk is handed to the decoder, so a peer
    that streams encoded bytes without end - chunked or with no
    ``Content-Length`` - is stopped at the fixed budget instead of feeding the
    decoding pipeline forever.
    """

    def __init__(self, limit: int | None = None) -> None:
        self.total = 0
        self._limit = limit

    def count(self, chunks: Iterable[bytes]) -> Iterator[bytes]:
        for chunk in chunks:
            self.total += len(chunk)
            if self._limit is not None and self.total > self._limit:
                raise ResponseTooLargeError(
                    f"encoded response exceeds the {self._limit}-byte limit"
                )
            yield chunk


def _strip_url_tail(value: str) -> str:
    """Fallback scrub for a URL that :func:`urlsplit` cannot parse."""
    head = re.split(r"[?#]", value, maxsplit=1)[0]
    return re.sub(r"//[^/@]*@", "//", head)


def redact_url(url: str) -> str:
    """Return ``url`` without userinfo, query string or fragment.

    This is the single redaction used by every message and log line of this
    plugin, so a password embedded in the URL or a signed query token can never
    be written back out.
    """
    try:
        parts = urlsplit(str(url))
        host = parts.hostname
        try:
            port = parts.port
        except ValueError:
            # A non-numeric port makes ``port`` raise; the host is still known.
            port = None
    except (AttributeError, ValueError):
        return _strip_url_tail(str(url))

    if not host:
        return _strip_url_tail(str(url))

    netloc = host if port is None else f"{host}:{port}"
    return urlunsplit((parts.scheme, netloc, parts.path, "", ""))


_URL_IN_TEXT = re.compile(r"[A-Za-z][A-Za-z0-9+.\-]*://[^\s'\"<>()\[\],;]+")


def redact_text(text: str) -> str:
    """Redact credentials/query/fragment from every URL inside ``text``.

    Used as the last boundary before exception text (or a traceback built from
    it) is handed to a logger, so a URL that reached a chained cause or a
    server-controlled header is still scrubbed.
    """
    if not text:
        return text
    return _URL_IN_TEXT.sub(lambda match: redact_url(match.group(0)), str(text))


def _scrub_exception(exc: BaseException | None) -> BaseException | None:
    """Strip credentials/query/fragment from an exception's own text.

    ``raise ... from exc`` keeps ``exc`` in the traceback, so the chained
    exception has to be scrubbed before it is re-raised.
    """
    if exc is None:
        return None
    try:
        if exc.args:
            exc.args = tuple(
                redact_text(arg) if isinstance(arg, str) else arg for arg in exc.args
            )
    except Exception:  # pragma: no cover - exotic exception objects
        pass
    return exc


class _AbortWatch:
    """End a blocking socket operation at the deadline or on cancellation.

    ``socket.settimeout`` only bounds the idle time *between* two reads, so a
    peer that drips one byte before every timeout can keep ``getresponse()``
    (status and header parsing) or the TLS handshake blocked indefinitely.
    Shutting the socket down from a helper thread makes every blocked socket
    call return, which is what lets the worker thread end and give its fetch
    slot back.
    """

    _POLL_SECONDS = 0.05

    def __init__(self, deadline: float, stop_event: Any) -> None:
        self._deadline = deadline
        self._stop_event = stop_event
        self._sock: socket.socket | None = None
        self._lock = threading.Lock()
        self._finished = threading.Event()
        self._triggered = threading.Event()
        self._thread = threading.Thread(
            target=self._watch, name="safe-fetch-abort", daemon=True
        )

    @property
    def triggered(self) -> bool:
        """``True`` when this watch (not the peer) ended the socket."""
        return self._triggered.is_set()

    def attach(self, sock: socket.socket) -> None:
        """Start watching ``sock``; call once, before its first blocking use."""
        with self._lock:
            self._sock = sock
        self._thread.start()

    def _watch(self) -> None:
        while not self._finished.wait(self._POLL_SECONDS):
            if self._stop_event is not None and self._stop_event.is_set():
                self._triggered.set()
                self._abort()
                return
            if time.monotonic() >= self._deadline:
                self._triggered.set()
                self._abort()
                return

    def _abort(self) -> None:
        with self._lock:
            sock = self._sock
        if sock is None:
            return
        try:
            sock.shutdown(socket.SHUT_RDWR)
        except OSError:
            pass
        try:
            sock.close()
        except OSError:
            pass

    def stop(self) -> None:
        """Stop watching; safe to call on every exit path."""
        self._finished.set()
        if self._thread.is_alive():
            self._thread.join(1.0)


@dataclasses.dataclass(frozen=True)
class Document:
    """A bounded, decoded HTTP response."""

    status_code: int
    url: str
    content_type: str
    text: str


@dataclasses.dataclass(frozen=True)
class _RawResponse:
    status_code: int
    headers: Any
    body: bytes
    charset: str | None


def is_public_address(value: str) -> bool:
    """Return ``True`` only for addresses that are globally routable."""
    try:
        address = ipaddress.ip_address(value)
    except ValueError:
        return False
    if address.version == 6 and address.ipv4_mapped is not None:
        # ::ffff:127.0.0.1 is loopback, not a public IPv6 address.
        address = address.ipv4_mapped
    if (
        address.is_private
        or address.is_loopback
        or address.is_link_local
        or address.is_multicast
        or address.is_reserved
        or address.is_unspecified
    ):
        return False
    return address.is_global


def resolve_public_addresses(
    host: str, port: int, deadline: float | None = None, stop_event: Any = None
) -> list[str]:
    """Resolve ``host`` and return its addresses, refusing non-public ones.

    With a ``deadline`` the lookup runs on the bounded resolver pool and this
    call gives up as soon as the deadline expires or ``stop_event`` is set: a
    resolver that never answers is another way to pin a fetch slot forever.
    Giving up does not free the resolution's budget slot - that happens only
    when ``getaddrinfo`` really ends - so a stalled resolver cannot be used to
    accumulate unbounded background work.
    """
    infos = _getaddrinfo(host, port, deadline, stop_event)

    addresses: list[str] = []
    for info in infos:
        address = info[4][0]
        if not is_public_address(address):
            raise BlockedTargetError(
                f"refusing non-public address {address!r} for host {host!r}"
            )
        if address not in addresses:
            addresses.append(address)
    if not addresses:
        raise BlockedTargetError(f"host {host!r} resolved to no address")
    return addresses


class _Resolution:
    """One in-flight ``getaddrinfo`` call and its result slot."""

    __slots__ = ("host", "port", "done", "infos", "error")

    def __init__(self, host: str, port: int) -> None:
        self.host = host
        self.port = port
        self.done = threading.Event()
        self.infos: list | None = None
        self.error: BaseException | None = None

    def execute(self) -> None:
        try:
            self.infos = socket.getaddrinfo(self.host, self.port, type=socket.SOCK_STREAM)
        except BaseException as exc:  # noqa: BLE001 - reported to the waiter
            self.error = exc


class _ResolverPool:
    """A fixed worker pool with a fixed, backpressured resolution budget.

    A resolution holds one budget slot until its ``getaddrinfo`` call really
    returns, even after the waiting caller has timed out or been cancelled, so
    an abandoned lookup can never free budget early or accumulate off the
    pool. The pool runs at most ``max_workers`` lookups at once, keeps at most
    ``capacity`` in flight (running plus queued), and refuses further work
    with :class:`ResolverBusyError` once the budget is exhausted instead of
    queueing without limit or starting another thread.
    """

    def __init__(self, max_workers: int, capacity: int) -> None:
        self._max_workers = max_workers
        self._capacity = capacity
        self._slots = threading.BoundedSemaphore(capacity)
        self._condition = threading.Condition()
        self._pending: collections.deque[_Resolution] = collections.deque()
        self._threads = 0
        self._running = 0
        self._in_flight = 0

    def stats(self) -> dict[str, int]:
        """Return a live snapshot of the pool, always within the fixed caps."""
        with self._condition:
            return {
                "max_workers": self._max_workers,
                "workers": self._threads,
                "capacity": self._capacity,
                "running": self._running,
                "queued": len(self._pending),
                "in_flight": self._in_flight,
            }

    def submit(self, host: str, port: int) -> "_Resolution":
        """Take a budget slot and queue ``host``; refuse when none is free."""
        if not self._slots.acquire(blocking=False):
            raise ResolverBusyError(
                "resolver budget exhausted; refusing another DNS lookup"
            )
        resolution = _Resolution(host, port)
        with self._condition:
            self._in_flight += 1
            self._pending.append(resolution)
            idle = self._threads - self._running
            if self._threads < self._max_workers and len(self._pending) > idle:
                self._threads += 1
                threading.Thread(
                    target=self._work, name="safe-fetch-dns", daemon=True
                ).start()
            self._condition.notify()
        return resolution

    def _work(self) -> None:
        while True:
            with self._condition:
                while not self._pending:
                    self._condition.wait()
                resolution = self._pending.popleft()
                self._running += 1
            try:
                resolution.execute()
            finally:
                # The slot is returned only now, when the resolver call has
                # really ended - never when a waiter gave up on it.
                with self._condition:
                    self._running -= 1
                    self._in_flight -= 1
                self._slots.release()
                resolution.done.set()


_RESOLVER_POOL = _ResolverPool(MAX_DNS_WORKERS, MAX_DNS_OUTSTANDING)


def resolver_stats() -> dict[str, int]:
    """Return the resolver pool's budget state (bounded by the module caps)."""
    return _RESOLVER_POOL.stats()


def _getaddrinfo(host: str, port: int, deadline: float | None, stop_event: Any) -> list:
    """Resolve ``host``, bounded by a deadline/stop flag when one is given.

    Without a ``deadline`` the lookup runs inline: it cannot outlive its caller
    and therefore needs no background budget. With a ``deadline`` it runs on
    the bounded resolver pool; a waiter that times out or is cancelled stops
    waiting but the resolution keeps its budget slot until it really ends, and
    further lookups are refused with :class:`ResolverBusyError` while the
    budget is exhausted.
    """
    if deadline is None:
        try:
            return socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
        except OSError as exc:
            raise BlockedTargetError(f"cannot resolve host {host!r}: {exc}") from exc

    resolution = _RESOLVER_POOL.submit(host, port)
    while True:
        if stop_event is not None and stop_event.is_set():
            raise FetchAbortedError("fetch cancelled while resolving host")
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise FetchAbortedError("fetch deadline exceeded while resolving host")
        if resolution.done.wait(min(0.05, remaining)):
            break

    error = resolution.error
    if error is not None:
        raise BlockedTargetError(
            f"cannot resolve host {host!r}: {redact_text(str(error))}"
        ) from _scrub_exception(error)
    return resolution.infos or []


def validate_url(url: str) -> tuple[str, str, int, str]:
    """Validate scheme/host/port and return ``(scheme, host, port, path)``."""
    try:
        parts = urlsplit(url)
    except ValueError as exc:
        raise BlockedTargetError(
            f"invalid URL: {redact_url(url)!r}"
        ) from _scrub_exception(exc)
    scheme = (parts.scheme or "").lower()
    if scheme not in _DEFAULT_PORTS:
        raise BlockedTargetError(f"refusing non-HTTP(S) URL: {redact_url(url)!r}")
    host = parts.hostname
    if not host:
        raise BlockedTargetError(f"URL has no host: {redact_url(url)!r}")
    try:
        port = parts.port or _DEFAULT_PORTS[scheme]
    except ValueError as exc:
        raise BlockedTargetError(
            f"invalid port in URL: {redact_url(url)!r}"
        ) from _scrub_exception(exc)
    path = parts.path or "/"
    if parts.query:
        path = f"{path}?{parts.query}"
    return scheme, host, port, path


def fetch_document(
    url: str,
    *,
    timeout: float = TOTAL_TIMEOUT,
    stop_event: Any = None,
    max_bytes: int = MAX_RESPONSE_BYTES,
    headers: Mapping[str, str] | None = None,
) -> Document:
    """Fetch ``url`` under the validation and size rules of this module.

    ``stop_event`` is an optional :class:`threading.Event`; when set, the fetch
    aborts at the next blocking step. ``timeout`` is the deadline for the whole
    operation, redirects included, and it is enforced on DNS, connect, the TLS
    handshake, status/header parsing and the body.
    """
    deadline = time.monotonic() + timeout
    current = url

    for _hop in range(MAX_REDIRECTS + 1):
        scheme, host, port, path = validate_url(current)
        _check_abort(stop_event, deadline)
        addresses = resolve_public_addresses(host, port, deadline, stop_event)
        _check_abort(stop_event, deadline)

        raw = _request_once(
            scheme, host, port, path, addresses, current, deadline, stop_event, max_bytes, headers
        )

        if raw.status_code in _REDIRECT_STATUSES:
            location = raw.headers.get("Location")
            if location:
                current = urljoin(current, location)
                continue

        return Document(
            status_code=raw.status_code,
            url=current,
            content_type=raw.headers.get("Content-Type", "") or "",
            text=_decode_text(raw.body, raw.charset),
        )

    raise TooManyRedirectsError(
        f"more than {MAX_REDIRECTS} redirects from {redact_url(url)!r}"
    )


def _check_abort(stop_event: Any, deadline: float) -> None:
    if stop_event is not None and stop_event.is_set():
        raise FetchAbortedError("fetch cancelled")
    if time.monotonic() >= deadline:
        raise FetchAbortedError("fetch deadline exceeded")


def _remaining(deadline: float) -> float:
    return max(0.1, deadline - time.monotonic())


def _request_once(
    scheme: str,
    host: str,
    port: int,
    path: str,
    addresses: list[str],
    url: str,
    deadline: float,
    stop_event: Any,
    max_bytes: int,
    headers: Mapping[str, str] | None,
) -> _RawResponse:
    sock = None
    last_error: OSError | None = None
    for address in addresses:
        # A stalled connect must not outlive the deadline or the cancellation.
        _check_abort(stop_event, deadline)
        try:
            sock = socket.create_connection(
                (address, port), timeout=min(CONNECT_TIMEOUT, _remaining(deadline))
            )
            break
        except OSError as exc:
            last_error = exc
    if sock is None:
        _check_abort(stop_event, deadline)
        raise FetchError(
            f"cannot connect to {redact_url(url)!r}: {redact_text(str(last_error))}"
        )

    connection = None
    watch = _AbortWatch(deadline, stop_event)
    try:
        sock.settimeout(min(READ_TIMEOUT, _remaining(deadline)))
        if scheme == "https":
            context = ssl.create_default_context()
            # The socket goes to the validated IP, the certificate and SNI stay
            # bound to the requested host name.  The handshake is deferred so
            # the abort watch can cover it as well.
            sock = context.wrap_socket(
                sock, server_hostname=host, do_handshake_on_connect=False
            )
        watch.attach(sock)
        if scheme == "https":
            sock.do_handshake()

        connection = http.client.HTTPConnection(host, port)
        connection.sock = sock
        connection.request("GET", path, headers=_request_headers(url, headers))
        response = connection.getresponse()

        status_code = response.status
        response_headers = response.headers
        if status_code in _REDIRECT_STATUSES and response_headers.get("Location"):
            # The body of a redirect is not needed; dropping the connection
            # discards it without reading it.
            _check_abort(stop_event, deadline)
            return _RawResponse(status_code, response_headers, b"", None)

        declared = response.length
        if declared is not None and declared > max_bytes:
            raise ResponseTooLargeError(
                f"Content-Length {declared} exceeds the {max_bytes}-byte limit for "
                f"{redact_url(url)!r}"
            )

        counter = _ByteCounter(max_bytes + _ENCODED_SLACK)
        decoder = _BodyDecoder(response_headers.get("Content-Encoding"), max_bytes)
        chunks = decoder.decode(counter.count(_raw_chunks(response, sock, stop_event, deadline)))
        body = collect_bounded(chunks, max_bytes)

        # Every exit path re-checks the abort state: a socket that the watch
        # shut down can read as a clean EOF, and such a partial body must never
        # be returned as if the response had been read completely.
        _check_abort(stop_event, deadline)
        if declared is not None and counter.total < declared:
            raise TruncatedResponseError(
                f"truncated response from {redact_url(url)!r}: "
                f"{counter.total} of {declared} bytes received"
            )
        charset = response_headers.get_content_charset()
        return _RawResponse(status_code, response_headers, body, charset)
    except FetchError:
        raise
    except (http.client.HTTPException, OSError, zlib.error) as exc:
        if watch.triggered:
            # The watch shut the socket down: the fetch was cancelled or ran
            # past its deadline, not a peer failure.
            raise FetchAbortedError(
                "fetch cancelled or deadline exceeded"
            ) from _scrub_exception(exc)
        raise FetchError(
            f"request to {redact_url(url)!r} failed: {redact_text(str(exc))}"
        ) from _scrub_exception(exc)
    finally:
        watch.stop()
        if connection is not None:
            connection.close()
        else:
            sock.close()


def _request_headers(url: str, extra: Mapping[str, str] | None) -> dict[str, str]:
    headers = dict(_REQUEST_DEFAULTS)
    if extra:
        headers.update(extra)
    parts = urlsplit(url)
    if parts.username:
        # Preserve the previous requests-based behaviour for URLs that carry
        # credentials, without consulting netrc or the environment.
        credentials = unquote(parts.username)
        if parts.password:
            credentials += ":" + unquote(parts.password)
        token = base64.b64encode(credentials.encode("utf-8")).decode("ascii")
        headers["Authorization"] = f"Basic {token}"
    return headers


def _raw_chunks(
    response: http.client.HTTPResponse, sock: socket.socket, stop_event: Any, deadline: float
) -> Iterator[bytes]:
    """Yield the raw (still encoded) body in bounded chunks."""
    while True:
        _check_abort(stop_event, deadline)
        sock.settimeout(min(READ_TIMEOUT, _remaining(deadline)))
        try:
            chunk = response.read1(_CHUNK_SIZE)
        except socket.timeout as exc:
            raise FetchAbortedError(f"response read timed out: {exc}") from exc
        if not chunk:
            # Reading a shut-down socket can surface as a clean EOF (the peer
            # just stops sending); only an explicit re-check tells "the body
            # ended" apart from "we aborted the read".
            _check_abort(stop_event, deadline)
            return
        yield chunk


def collect_bounded(chunks: Iterable[bytes], limit: int) -> bytes:
    """Join ``chunks`` while refusing to buffer more than ``limit`` bytes.

    The check happens *before* a chunk is appended and before the next chunk is
    pulled, so an oversized body is detected without reading it all.
    """
    buffer = bytearray()
    for chunk in chunks:
        if not chunk:
            continue
        if len(buffer) + len(chunk) > limit:
            raise ResponseTooLargeError(f"response exceeds the {limit}-byte limit")
        buffer.extend(chunk)
    return bytes(buffer)


class _BodyDecoder:
    """Streaming gzip/deflate decoder with a cap on the inflated size.

    Exactly one compressed member is supported.  The decoder tracks zlib's
    ``eof`` flag and refuses to feed an end-of-stream decompressor again, so
    bytes that follow the member are neither buffered in ``unused_data`` nor
    inflated: they raise :class:`TrailingDataError` instead.  A stream that
    stops before its end marker raises
    :class:`IncompleteCompressedStreamError`, because only a complete stream
    can have its gzip CRC32 / zlib Adler-32 (and length) validated.
    """

    def __init__(self, encoding: str | None, limit: int) -> None:
        self.encoding = (encoding or "").strip().lower()
        self._limit = limit
        self._done = False
        if self.encoding in ("", "identity"):
            self._inflater = None
        elif self.encoding == "gzip":
            self._inflater = zlib.decompressobj(16 + zlib.MAX_WBITS)
        elif self.encoding == "deflate":
            self._inflater = zlib.decompressobj()
        else:
            raise FetchError(f"unsupported Content-Encoding: {self.encoding!r}")

    def decode(self, chunks: Iterable[bytes]) -> Iterator[bytes]:
        if self._inflater is None:
            yield from chunks
            return
        for chunk in chunks:
            if not chunk:
                continue
            if self._done:
                # The member already ended; do not feed the EOF'd decompressor
                # the trailing bytes and do not let them reach the decoder's
                # internal buffer.
                raise TrailingDataError(
                    "compressed response has data after the end of its stream"
                )
            out = self._inflater.decompress(chunk, self._limit)
            if self._inflater.unconsumed_tail:
                # More inflated bytes are pending than the cap allows: stop
                # before the decompression bomb can grow.
                raise ResponseTooLargeError(
                    f"decompressed response exceeds the {self._limit}-byte limit"
                )
            if out:
                yield out
            if self._inflater.eof:
                if self._inflater.unused_data:
                    # The stream ended inside this chunk and bytes follow it;
                    # refuse them instead of retaining them in ``unused_data``.
                    raise TrailingDataError(
                        "compressed response has data after the end of its stream"
                    )
                self._done = True
        if not self._done:
            raise IncompleteCompressedStreamError(
                "compressed response ended before the end of its stream"
            )
        tail = self._inflater.flush()
        if tail:
            yield tail


def _decode_text(body: bytes, charset: str | None) -> str:
    encoding = charset or "utf-8"
    try:
        return body.decode(encoding, errors="replace")
    except LookupError:
        return body.decode("utf-8", errors="replace")