diff --git a/deploy/docker/README.md b/deploy/docker/README.md
index 7ce0dcd00..b9e88e165 100644
--- a/deploy/docker/README.md
+++ b/deploy/docker/README.md
@@ -116,6 +116,102 @@ EOL
 
 > The server will be available at `http://localhost:11235`. Visit `/playground` to access the interactive testing interface.
 
+*   **Behind a corporate proxy:** if the host reaches the internet only through
+    an HTTP proxy, set the standard `HTTP_PROXY` / `HTTPS_PROXY` / `NO_PROXY`
+    env vars (Docker's `proxies` config injects them automatically) — the
+    server's egress proxy chains through it while keeping its SSRF protections
+    (the upstream is asked to CONNECT to an already-validated, pinned IP).
+    `CRAWL4AI_UPSTREAM_PROXY` overrides the env vars. Basic auth via
+    `http://user:pass@proxy:port` is supported; for NTLM/Kerberos proxies,
+    front them with a local translator (e.g. `cntlm`, `px`) and point
+    `CRAWL4AI_UPSTREAM_PROXY` at it. Containers with no DNS at all are not yet
+    supported.
+
+*   **Only some destinations belong to the upstream:** chaining is otherwise
+    all-or-nothing — with an upstream set, every crawl goes through it, and
+    `NO_PROXY` only subtracts exceptions from that. A deployment that has to
+    serve both routes cannot express itself that way: the internal sites need
+    the proxy, the public ones must not use it, and the public set cannot be
+    enumerated in `NO_PROXY`. Sending public crawls through the upstream anyway
+    is not merely wasteful — an upstream that authenticates or authorises per
+    destination will refuse them, and the fetch is attributed to the upstream's
+    network rather than yours.
+
+    `CRAWL4AI_UPSTREAM_PROXY_ONLY_SUFFIXES` inverts the rule for the names it
+    lists (comma-separated, same form as above): those are chained through the
+    upstream, everything else dials direct. Leave it unset — the default — and
+    behaviour is exactly as before, with every target chained.
+
+    It composes with `CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES` rather than
+    overriding it: a name may be handed over unresolved only if it is also
+    allowed to use the upstream, so neither list can widen what the other
+    permits. Matching is the same normalised, label-boundary comparison, a
+    wildcard entry is ignored with a warning, and an IP literal never matches a
+    suffix, so a pinned address is dialled direct. Nothing here relaxes the
+    pin: a target that is not chained is resolved and pinned exactly as it is
+    today, so the SSRF and rebinding protections apply to strictly more traffic
+    than before, never less.
+
+*   **Upstreams that must resolve the name themselves:** some upstreams front a
+    network whose hostnames do not resolve outside it, or enforce hostname ACLs
+    and so refuse CONNECT-to-an-IP. Those need the name rather than a pinned
+    address. List the suffixes to handle that way in
+    `CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES` (comma-separated, e.g.
+    `.corp.example.com,.internal`); matching names are sent to the upstream
+    unresolved, and the upstream then decides where the connection lands.
+
+    It is an allowlist rather than a switch, and it is a deliberate trust
+    decision: for those names you rely on the upstream instead of the built-in
+    pin. Everything outside the list — any other hostname, any IP literal, and
+    any deployment that leaves the variable unset — keeps resolve-and-pin with
+    its SSRF and rebinding protections unchanged. IP literals never qualify, so
+    listing a suffix cannot expose link-local or metadata addresses.
+
+    The same list exempts those names from the entry-point URL check, which runs
+    before the browser starts; without that they would be rejected there and
+    never reach the proxy. Use it instead of `CRAWL4AI_ALLOW_INTERNAL_URLS`,
+    which disables that check for every destination rather than for the names
+    you nominated.
+
+    Matching is on the DNS label boundary, so `.corp.example` covers
+    `wiki.corp.example` but never `corp.example.attacker.com` or
+    `notcorp.example`. Names are compared in normalised form — case-folded, root
+    dot stripped, IDNA-encoded — so the spellings of one name cannot disagree
+    with the list or with each other, and the normalised name is also the one sent
+    to the upstream, so the name that was authorised is the name on the wire. A
+    wildcard entry is ignored with a warning
+    rather than honoured: delegating every name would turn any caller-supplied
+    URL into a lookup performed by the upstream, which is the one shape this
+    must not allow.
+
+    Delegated names are reachable on ports 80 and 443 only. On the pinned path
+    the port is unconstrained but the address must be global, so a non-web port
+    still only reaches the public internet; passthrough gives up exactly that
+    address check, which leaves the port as the only thing still narrowing where
+    a listed name can land. Without it a crawl of a listed suffix could reach an
+    internal database or admin port — `:9200`, `:5432`, `:2375` — rather than a
+    web server. Set `CRAWL4AI_UPSTREAM_PROXY_DNS_PORTS` (e.g. `80,443,8443`) if
+    you front an internal site elsewhere; the list replaces the default rather
+    than extending it, and anything outside it falls back to resolve-and-pin.
+
+    **What you are trusting.** For a delegated name there is no pin, so the
+    rebinding and non-global checks do not apply to it — that is the point of
+    the mode, and it is worth being explicit about the consequence. A subdomain
+    under a listed suffix reaches whatever its DNS resolves to, including a
+    private or link-local address, because we neither resolve nor pin it. The
+    connection is still only ever handed to your configured upstream and never
+    dialed directly, so this cannot reach the crawler's own network — but it can
+    reach whatever the upstream is willing to reach. List suffixes whose DNS you
+    control, keep them as narrow as possible, and prefer an upstream that
+    enforces its own destination policy.
+
+    Note this is narrower than the escape hatch that already exists:
+    `CRAWL4AI_ALLOW_INTERNAL_URLS=true` drops the entry-point check for every
+    destination and still leaves the proxy pinning, so an operator in this
+    situation has had to reach for it anyway. The suffix list confines the same
+    relaxation to named hosts, keeps IP literals blocked, and never dials a
+    target directly.
+
 #### 4. Stopping the Container
 
 ```bash
diff --git a/deploy/docker/egress_proxy.py b/deploy/docker/egress_proxy.py
index 4cdb56bd1..e11b79a24 100644
--- a/deploy/docker/egress_proxy.py
+++ b/deploy/docker/egress_proxy.py
@@ -13,13 +13,29 @@
 against the real host - no MITM).
 
 Bound to 127.0.0.1 on an ephemeral port; started at server boot.
+
+If HTTP_PROXY/HTTPS_PROXY (or CRAWL4AI_UPSTREAM_PROXY) is set, we still
+resolve-and-pin locally but dial via the upstream proxy, asking it to CONNECT
+to the PINNED IP — never the hostname — so the rebinding guarantee holds.
+NO_PROXY bypasses it; with no proxy env set, behavior is unchanged.
+
+CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES is the one exception, and it is an
+allowlist rather than a switch: names matching a listed suffix are handed to the
+upstream unresolved, because some upstreams front a network whose names do not
+resolve here at all. For those names the pin is the upstream's responsibility
+instead of ours. Everything else — every other name, every IP literal, and every
+deployment that leaves the variable unset — keeps resolve-and-pin unchanged.
 """
 
 from __future__ import annotations
 
 import asyncio
+import base64
+import ipaddress
 import logging
-from urllib.parse import urlsplit
+import os
+import re
+from urllib.parse import unquote, urlsplit
 
 from egress_broker import EgressBlocked, resolve_and_pin
 
@@ -31,6 +47,295 @@
 _MAX_HEADER_BYTES = 64 * 1024
 
 
+def _env(*names: str) -> str:
+    return next((os.environ[n] for n in names if os.environ.get(n)), "")
+
+
+def upstream_proxy(scheme: str = "https"):
+    """(host, port, auth_header_bytes|None) of the upstream proxy, or None.
+
+    Read per-call (not at import) so operators and tests see env changes.
+    The target scheme picks HTTP(S)_PROXY per convention, falling back to
+    the other pair when only one is set.
+    """
+    order = ("HTTP_PROXY", "http_proxy", "HTTPS_PROXY", "https_proxy") if scheme == "http" \
+        else ("HTTPS_PROXY", "https_proxy", "HTTP_PROXY", "http_proxy")
+    raw = _env("CRAWL4AI_UPSTREAM_PROXY", *order).strip()
+    if not raw:
+        return None
+    sp = urlsplit(raw if "://" in raw else "http://" + raw)
+    if not sp.hostname:
+        return None
+    auth = None
+    if sp.username:
+        cred = f"{unquote(sp.username)}:{unquote(sp.password or '')}".encode("utf-8")
+        auth = b"Proxy-Authorization: Basic " + base64.b64encode(cred) + b"\r\n"
+    return sp.hostname, sp.port or 80, auth
+
+
+def _no_proxy_match(host: str, ip: str) -> bool:
+    """True if NO_PROXY says this target must bypass the upstream proxy."""
+    entries = [e.strip() for e in _env("NO_PROXY", "no_proxy").split(",") if e.strip()]
+    for entry in entries:
+        if entry == "*":
+            return True
+        try:
+            if ipaddress.ip_address(ip) in ipaddress.ip_network(entry, strict=False):
+                return True
+            continue
+        except ValueError:
+            pass
+        suffix = entry.lower().lstrip(".")
+        low = host.lower()
+        if low == suffix or low.endswith("." + suffix):
+            return True
+    return False
+
+
+def upstream_only_suffixes() -> list:
+    """Suffixes that are the ONLY targets routed through the upstream proxy.
+
+    Empty (the default) keeps the existing rule: with an upstream set every
+    target is chained through it and NO_PROXY carves out exceptions. That fits a
+    site whose whole egress leaves via one corporate proxy, but it makes a single
+    deployment unable to serve two egress routes at once — a public crawl is
+    tunnelled through the proxy as well, where it may be refused outright or
+    attributed to the wrong network.
+
+    Naming suffixes here inverts the rule for those names only: they are chained,
+    everything else dials direct. It is an allowlist for the same reason
+    CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES is one, and the two compose: a name may
+    be delegated unresolved only if it is also allowed to use the upstream, so
+    neither list can widen what the other permits.
+    """
+    raw = _env("CRAWL4AI_UPSTREAM_PROXY_ONLY_SUFFIXES")
+    out = []
+    for entry in raw.split(","):
+        entry = entry.strip().lstrip(".")
+        if not entry:
+            continue
+        if "*" in entry:
+            logger.warning(
+                "ignoring wildcard entry %r in CRAWL4AI_UPSTREAM_PROXY_ONLY_SUFFIXES; "
+                "list explicit suffixes instead", entry,
+            )
+            continue
+        normalized = normalize_host(entry)
+        if not normalized:
+            logger.warning(
+                "ignoring invalid suffix %r in CRAWL4AI_UPSTREAM_PROXY_ONLY_SUFFIXES", entry,
+            )
+            continue
+        out.append(normalized)
+    return out
+
+
+def _upstream_allows(host: str) -> bool:
+    """True if `host` may use the upstream proxy at all.
+
+    With no allowlist configured every host may, which is the existing rule. The
+    comparison is on the DNS label boundary, so "corp.example" matches
+    "wiki.corp.example" but never "corp.example.attacker.com".
+    """
+    only = upstream_only_suffixes()
+    if not only:
+        return True
+    low = normalize_host(host)
+    if not low:
+        return False
+    return any(low == s or low.endswith("." + s) for s in only)
+
+
+def _use_upstream(pin):
+    """The upstream (host, port, auth) to route `pin` through, or None for direct."""
+    up = upstream_proxy(pin.scheme)
+    if up is None or _no_proxy_match(pin.host, pin.ip):
+        return None
+    if not _upstream_allows(pin.host):
+        return None
+    return up
+
+
+_MAX_HOSTNAME = 253
+_LABEL = re.compile(r"^[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?$")
+
+
+def normalize_host(host: str) -> str:
+    """Lowercase, strip the root dot, and IDNA-encode; "" if not a valid name.
+
+    Comparing raw input against the allowlist would let equivalent spellings of
+    the same name disagree — a trailing root dot, uppercase, or a unicode form
+    of an ASCII label. Anything that is not a syntactically valid hostname
+    returns "" so it can never match and simply keeps the pinned path.
+    """
+    host = (host or "").strip().rstrip(".")
+    if not host or len(host) > _MAX_HOSTNAME:
+        return ""
+    try:
+        host = host.encode("idna").decode("ascii")
+    except (UnicodeError, UnicodeDecodeError):
+        # Already ASCII, or not encodable; fall back and let the label check rule.
+        try:
+            host.encode("ascii")
+        except UnicodeEncodeError:
+            return ""
+    # Re-check: the A-label form of a unicode name is usually longer than the
+    # name we measured above, so the first check does not bound this one.
+    if len(host) > _MAX_HOSTNAME:
+        return ""
+    host = host.lower()
+    labels = host.split(".")
+    if not all(_LABEL.match(label) for label in labels):
+        return ""
+    return host
+
+
+def passthrough_suffixes() -> list:
+    """Hostname suffixes the operator allows the upstream to resolve.
+
+    Empty (the default) disables passthrough entirely. This is deliberately an
+    allowlist rather than a boolean: the pin is what stops a caller-supplied URL
+    from reaching link-local or private space, so it is only given up for the
+    exact names an operator names, never wholesale.
+
+    A bare "*" is rejected rather than honoured. Delegating every name would
+    turn any attacker-supplied URL into a lookup performed by the upstream, which
+    is the one shape this must not allow; an operator who genuinely wants that
+    already has CRAWL4AI_ALLOW_INTERNAL_URLS.
+    """
+    raw = _env("CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES")
+    out = []
+    for entry in raw.split(","):
+        entry = entry.strip().lstrip(".")
+        if not entry:
+            continue
+        if entry == "*" or "*" in entry:
+            logger.warning(
+                "ignoring wildcard entry %r in CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES; "
+                "list explicit suffixes instead", entry,
+            )
+            continue
+        normalized = normalize_host(entry)
+        if not normalized:
+            logger.warning(
+                "ignoring invalid suffix %r in CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES", entry,
+            )
+            continue
+        out.append(normalized)
+    return out
+
+
+_DEFAULT_PASSTHROUGH_PORTS = (80, 443)
+
+
+def passthrough_ports() -> frozenset:
+    """Ports a delegated name may be reached on; defaults to the web ports.
+
+    On the pinned path the port is unconstrained but the address must be global,
+    so a non-web port can still only reach the public internet. Passthrough
+    gives up exactly that address check, which leaves the port as the only thing
+    still narrowing where a delegated name can land — unconstrained, a crawl of
+    a listed suffix could reach an internal database or admin port rather than a
+    web server. Defaulting to 80 and 443 keeps the mode doing what it exists
+    for: crawling.
+
+    An operator fronting an internal site on another port can list it, and an
+    explicit list replaces the default rather than extending it. Unset (or
+    blank) means the default; a list that is set but yields no usable port
+    disables passthrough rather than opening it up.
+    """
+    raw = _env("CRAWL4AI_UPSTREAM_PROXY_DNS_PORTS").strip()
+    if not raw:
+        return frozenset(_DEFAULT_PASSTHROUGH_PORTS)
+    out = set()
+    for entry in raw.split(","):
+        entry = entry.strip()
+        if not entry:
+            continue
+        try:
+            port = int(entry)
+        except ValueError:
+            logger.warning(
+                "ignoring non-numeric port %r in CRAWL4AI_UPSTREAM_PROXY_DNS_PORTS", entry,
+            )
+            continue
+        if not 0 < port < 65536:
+            logger.warning(
+                "ignoring out-of-range port %d in CRAWL4AI_UPSTREAM_PROXY_DNS_PORTS", port,
+            )
+            continue
+        out.add(port)
+    return frozenset(out)
+
+
+def _no_proxy_host_match(host: str) -> bool:
+    """NO_PROXY check by hostname alone, for targets we deliberately do not resolve.
+
+    The CIDR entries in _no_proxy_match cannot apply here: opting into
+    passthrough means we never learn an IP for this target.
+    """
+    for entry in [e.strip() for e in _env("NO_PROXY", "no_proxy").split(",") if e.strip()]:
+        if entry == "*":
+            return True
+        suffix = entry.lower().lstrip(".")
+        low = host.lower()
+        if low == suffix or low.endswith("." + suffix):
+            return True
+    return False
+
+
+def _passthrough_upstream(scheme: str, host: str, port: int):
+    """(upstream, authorised name) to hand `host` to unresolved, or None.
+
+    The normalised name is returned rather than recomputed by each caller so the
+    name that goes on the wire is necessarily the one the allowlist approved.
+    Deriving it twice would let the check and the connection disagree, which is
+    the shape of bug this whole path has to avoid.
+
+    Returning None keeps the caller on resolve-and-pin, so a deployment that has
+    not configured any suffix behaves byte-identically to today. It is also how
+    a rejected target fails closed: the caller then resolves and pins it like
+    any other, which is what blocks it — nothing skips both checks.
+
+    Five conditions, all required:
+      - the operator listed a matching suffix (an allowlist, not a switch);
+      - CRAWL4AI_UPSTREAM_PROXY_ONLY_SUFFIXES, if set, also allows the name —
+        a name may not be delegated to an upstream it is not allowed to use;
+      - the target is a name, not an IP literal — an address carries no DNS to
+        defer, so letting one through would hand the upstream 169.254.169.254
+        verbatim and give up the non-global rule for nothing;
+      - the port is one passthrough is allowed to reach;
+      - an upstream exists and NO_PROXY does not exempt the host.
+    """
+    suffixes = passthrough_suffixes()
+    if not suffixes:
+        return None
+    try:
+        ipaddress.ip_address(host.strip().rstrip("."))
+        return None
+    except ValueError:
+        pass
+    low = normalize_host(host)
+    if not low:
+        return None
+    # Compare on the DNS label boundary, so "corp.example" matches
+    # "wiki.corp.example" but never "corp.example.attacker.com" or "notcorp.example".
+    if not any(low == s or low.endswith("." + s) for s in suffixes):
+        return None
+    if port not in passthrough_ports():
+        return None
+    up = upstream_proxy(scheme)
+    if up is None or _no_proxy_host_match(low):
+        return None
+    if not _upstream_allows(low):
+        return None
+    return up, low
+
+
+def _bracket(ip: str) -> str:
+    return f"[{ip}]" if ":" in ip else ip
+
+
 class PinningProxy:
     """Async HTTP forward-proxy that connects only to pinned, global IPs."""
 
@@ -52,6 +357,12 @@ async def start(self) -> str:
         sock = self._server.sockets[0]
         self.bound_host, self.bound_port = sock.getsockname()[:2]
         logger.info("egress pinning proxy listening on %s", self.url)
+        up = upstream_proxy()
+        if up is not None:
+            logger.info(
+                "egress pinning proxy chaining through upstream proxy %s:%s",
+                up[0], up[1],
+            )
         return self.url
 
     async def stop(self) -> None:
@@ -91,19 +402,30 @@ async def _handle_connect(self, target, client_reader, client_writer):
         if not host or not port_s.isdigit():
             await self._reply(client_writer, _BAD)
             return
-        try:
-            pin = resolve_and_pin(f"https://{host}:{port_s}")
-        except EgressBlocked:
-            await self._reply(client_writer, _BLOCKED)
-            return
+        # Opt-in: hand the hostname to the upstream and let it resolve. Decided
+        # before resolve_and_pin, because the point of the mode is that this name
+        # may not resolve here at all, or may resolve to an address the pin
+        # refuses. Without the opt-in this is None and nothing below changes.
+        passthrough = _passthrough_upstream("https", host, int(port_s))
+
+        if passthrough is None:
+            try:
+                pin = resolve_and_pin(f"https://{host}:{port_s}")
+            except EgressBlocked:
+                await self._reply(client_writer, _BLOCKED)
+                return
 
         # Drain the rest of the client's CONNECT headers.
         await self._drain_headers(client_reader)
 
         try:
-            up_reader, up_writer = await asyncio.wait_for(
-                asyncio.open_connection(pin.ip, int(port_s)), timeout=30
-            )
+            if passthrough is not None:
+                up, delegated_host = passthrough
+                up_reader, up_writer = await self._dial_hostname(
+                    up, delegated_host, int(port_s)
+                )
+            else:
+                up_reader, up_writer = await self._dial(pin, int(port_s))
         except Exception:
             await self._reply(client_writer, _BLOCKED)
             return
@@ -119,26 +441,57 @@ async def _handle_absolute(self, method, target, request_line, client_reader, cl
             await self._reply(client_writer, _BAD)
             return
         port = sp.port or 80
-        try:
-            pin = resolve_and_pin(f"http://{sp.hostname}:{port}")
-        except EgressBlocked:
-            await self._reply(client_writer, _BLOCKED)
-            return
+        # Same opt-in as the CONNECT path, decided before we try to resolve.
+        passthrough = _passthrough_upstream("http", sp.hostname, port)
+        pin = None
+        if passthrough is None:
+            try:
+                pin = resolve_and_pin(f"http://{sp.hostname}:{port}")
+            except EgressBlocked:
+                await self._reply(client_writer, _BLOCKED)
+                return
 
         headers = await self._read_headers(client_reader)
         path = sp.path or "/"
         if sp.query:
             path += "?" + sp.query
+        delegated_host = passthrough[1] if passthrough is not None else None
+        upstream = passthrough[0] if passthrough is not None else _use_upstream(pin)
+        dst = (upstream[0], upstream[1]) if upstream else (pin.ip, port)
         try:
             up_reader, up_writer = await asyncio.wait_for(
-                asyncio.open_connection(pin.ip, port), timeout=30
+                asyncio.open_connection(*dst), timeout=30
             )
         except Exception:
             await self._reply(client_writer, _BLOCKED)
             return
-        # Re-issue in origin form with Host preserved.
-        out = f"{method} {path} HTTP/1.1\r\n".encode("latin-1")
-        out += b"Host: " + sp.hostname.encode("latin-1")
+        # Re-issue with Host preserved: origin form when dialing the pinned IP
+        # directly, absolute form against the pinned IP when going through the
+        # upstream proxy (which then needs no DNS lookup of its own).
+        if upstream is None:
+            out = f"{method} {path} HTTP/1.1\r\n".encode("latin-1")
+        else:
+            # Absolute form to the upstream. Against a delegated name send the
+            # name the allowlist approved and let the upstream resolve it;
+            # otherwise send the pinned IP, so the upstream needs no lookup.
+            origin = (
+                f"{delegated_host}:{port}" if passthrough is not None
+                else f"{_bracket(pin.ip)}:{port}"
+            )
+            out = f"{method} http://{origin}{path} HTTP/1.1\r\n".encode("latin-1")
+            if upstream[2]:
+                out += upstream[2]
+            # One validated request per upstream connection: only this first
+            # request is pinned/rewritten, so force close to keep a reused
+            # client connection from smuggling unvalidated requests upstream.
+            # A delegated name needs this just as much: the suffix and port were
+            # checked for this request only.
+            headers = b"".join(
+                ln + b"\r\n" for ln in headers.split(b"\r\n")
+                if ln and not ln.lower().startswith(b"connection:")
+            ) + b"Connection: close\r\n"
+        host_header = delegated_host if passthrough is not None else sp.hostname
+        out += b"Host: " + host_header.encode("latin-1")
         if sp.port:
             out += f":{sp.port}".encode("latin-1")
         out += b"\r\n" + headers + b"\r\n"
@@ -147,6 +500,77 @@ async def _handle_absolute(self, method, target, request_line, client_reader, cl
         await self._splice(client_reader, client_writer, up_reader, up_writer)
 
     # ─────────────────────────── helpers ───────────────────────────
+    async def _dial_hostname(self, upstream, host: str, port: int):
+        """CONNECT to the upstream by hostname, letting it resolve.
+
+        Only reachable once the operator opted in. `host` is the name the
+        allowlist approved, already normalised to its A-label form, so it goes
+        out as-is for a proxy that fronts a private network, or enforces
+        hostname ACLs, to route; we perform no lookup of our own, which is the
+        point — the name may not resolve on this side at all. ASCII is the wire
+        form of a hostname and normalisation is what guarantees it here.
+        """
+        p_host, p_port, auth = upstream
+        reader, writer = await asyncio.wait_for(
+            asyncio.open_connection(p_host, p_port), timeout=30
+        )
+        try:
+            dst = f"{host}:{port}"
+            req = f"CONNECT {dst} HTTP/1.1\r\nHost: {dst}\r\n".encode("ascii")
+            if auth:
+                req += auth
+            req += b"\r\n"
+            writer.write(req)
+            await writer.drain()
+            status = await asyncio.wait_for(reader.readline(), timeout=30)
+            parts = status.split()
+            if len(parts) < 2 or parts[1] != b"200":
+                logger.warning(
+                    "upstream proxy refused CONNECT %s: %s", dst, status.strip()
+                )
+                raise EgressBlocked("upstream proxy refused CONNECT")
+            while True:
+                line = await asyncio.wait_for(reader.readline(), timeout=30)
+                if line in (b"\r\n", b"\n", b""):
+                    break
+            return reader, writer
+        except Exception:
+            writer.close()
+            raise
+
+    async def _dial(self, pin, port: int):
+        """Open a byte pipe to the pinned IP: direct, or tunneled through the
+        upstream proxy via CONNECT-to-the-pinned-IP (no upstream DNS lookup)."""
+        upstream = _use_upstream(pin)
+        if upstream is None:
+            return await asyncio.wait_for(
+                asyncio.open_connection(pin.ip, port), timeout=30
+            )
+        p_host, p_port, auth = upstream
+        reader, writer = await asyncio.wait_for(
+            asyncio.open_connection(p_host, p_port), timeout=30
+        )
+        try:
+            dst = f"{_bracket(pin.ip)}:{port}"
+            req = f"CONNECT {dst} HTTP/1.1\r\nHost: {dst}\r\n".encode("latin-1")
+            if auth:
+                req += auth
+            req += b"\r\n"
+            writer.write(req)
+            await writer.drain()
+            status = await asyncio.wait_for(reader.readline(), timeout=30)
+            parts = status.split()
+            if len(parts) < 2 or parts[1] != b"200":
+                logger.warning("upstream proxy refused CONNECT: %r", status[:64])
+                raise ConnectionError("upstream proxy refused CONNECT")
+            # Drain the upstream's response headers so none of them leak into
+            # the tunneled byte stream.
+            await self._drain_headers(reader)
+        except Exception:
+            await self._safe_close(writer)
+            raise
+        return reader, writer
+
     async def _drain_headers(self, reader):
         read = 0
         while True:
diff --git a/deploy/docker/tests/test_security_egress_proxy.py b/deploy/docker/tests/test_security_egress_proxy.py
index 03ebae1ea..b5178f04a 100644
--- a/deploy/docker/tests/test_security_egress_proxy.py
+++ b/deploy/docker/tests/test_security_egress_proxy.py
@@ -20,6 +20,19 @@
 
 pytestmark = pytest.mark.posture
 
+_PROXY_ENV = (
+    "CRAWL4AI_UPSTREAM_PROXY", "HTTP_PROXY", "http_proxy",
+    "HTTPS_PROXY", "https_proxy", "NO_PROXY", "no_proxy",
+    "CRAWL4AI_UPSTREAM_PROXY_ONLY_SUFFIXES",
+)
+
+
+@pytest.fixture(autouse=True)
+def _clear_proxy_env(monkeypatch):
+    # Keep the suite deterministic on dev machines that sit behind a proxy.
+    for name in _PROXY_ENV:
+        monkeypatch.delenv(name, raising=False)
+
 
 async def _fake_upstream():
     async def handle(reader, writer):
@@ -121,6 +134,150 @@ async def test_malformed_connect_400(self):
             await proxy.stop()
 
 
+async def _fake_corporate_proxy(seen):
+    """Minimal HTTP proxy: records the CONNECT request line, replies 200, then
+    answers any tunneled bytes with TUNNEL-OK."""
+    async def handle(reader, writer):
+        line = await reader.readline()
+        seen.append(line)
+        while True:  # drain CONNECT headers
+            h = await reader.readline()
+            if h in (b"\r\n", b"\n", b""):
+                break
+        writer.write(b"HTTP/1.1 200 Connection established\r\nVia: fake\r\n\r\n")
+        await writer.drain()
+        await reader.read(65536)
+        writer.write(b"TUNNEL-OK")
+        await writer.drain()
+        writer.close()
+    server = await asyncio.start_server(handle, "127.0.0.1", 0)
+    return server, server.sockets[0].getsockname()[1]
+
+
+@pytest.mark.asyncio
+class TestUpstreamChaining:
+    async def test_chained_connect_pins_ip_and_blocks_before_upstream(self, monkeypatch):
+        """The chained-CONNECT security contract: the upstream receives the
+        PINNED IP (never a hostname to resolve), its response headers do not
+        leak into the tunnel, and a blocked target produces an opaque 403
+        with zero upstream traffic."""
+        seen = []
+        corp, corp_port = await _fake_corporate_proxy(seen)
+        monkeypatch.setenv("HTTPS_PROXY", f"http://127.0.0.1:{corp_port}")
+
+        def fake_pin(url):
+            if "internal.example" in url:
+                raise EgressBlocked()
+            return PinnedTarget("https", "good.example", 443, "203.0.113.7")
+        monkeypatch.setattr(egress_proxy, "resolve_and_pin", fake_pin)
+
+        proxy = PinningProxy()
+        await proxy.start()
+        try:
+            r, w = await asyncio.open_connection(proxy.bound_host, proxy.bound_port)
+            w.write(b"CONNECT good.example:443 HTTP/1.1\r\n\r\n")
+            await w.drain()
+            status = await asyncio.wait_for(r.readline(), timeout=5)
+            assert b"200" in status
+            await r.readline()  # blank line after the 200
+            w.write(b"hello")
+            await w.drain()
+            body = await asyncio.wait_for(r.read(100), timeout=5)
+            # Upstream's Via header must NOT leak into the tunnel.
+            assert body == b"TUNNEL-OK"
+            w.close()
+
+            r, w = await asyncio.open_connection(proxy.bound_host, proxy.bound_port)
+            w.write(b"CONNECT internal.example:443 HTTP/1.1\r\n\r\n")
+            await w.drain()
+            status = await asyncio.wait_for(r.readline(), timeout=5)
+            assert b"403" in status
+            w.close()
+        finally:
+            await proxy.stop()
+            corp.close()
+        # The upstream saw ONLY the pinned IP of the allowed target.
+        assert seen == [b"CONNECT 203.0.113.7:443 HTTP/1.1\r\n"]
+
+    async def test_chained_plain_http_pinned_absolute_form_no_smuggling(self, monkeypatch):
+        """Plain HTTP via upstream: the request is re-issued in absolute form
+        against the PINNED IP (no name for the upstream to resolve), carries
+        Connection: close, and a reused client connection cannot smuggle a
+        second, unvalidated request upstream."""
+        lines = []
+
+        async def handle(reader, writer):
+            req = b""
+            while b"\r\n\r\n" not in req:
+                req += await reader.read(4096)
+            lines.append(req)
+            writer.write(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nhi")
+            await writer.drain()
+            writer.close()
+        corp = await asyncio.start_server(handle, "127.0.0.1", 0)
+        corp_port = corp.sockets[0].getsockname()[1]
+        monkeypatch.setenv("HTTP_PROXY", f"http://127.0.0.1:{corp_port}")
+
+        def fake_pin(url):
+            return PinnedTarget("http", "plain.example", 80, "203.0.113.7")
+        monkeypatch.setattr(egress_proxy, "resolve_and_pin", fake_pin)
+
+        proxy = PinningProxy()
+        await proxy.start()
+        try:
+            r, w = await asyncio.open_connection(proxy.bound_host, proxy.bound_port)
+            w.write(b"GET http://plain.example/ HTTP/1.1\r\n"
+                    b"Host: plain.example\r\nConnection: keep-alive\r\n\r\n")
+            await w.drain()
+            first = await asyncio.wait_for(r.read(200), timeout=5)
+            assert b"200" in first
+            # Attempt to smuggle an unvalidated request on the same connection.
+            w.write(b"GET http://rebind.evil/ HTTP/1.1\r\nHost: rebind.evil\r\n\r\n")
+            await w.drain()
+            leftover = await asyncio.wait_for(r.read(200), timeout=5)
+            assert leftover == b""  # upstream closed; nothing came back
+            w.close()
+        finally:
+            await proxy.stop()
+            corp.close()
+        sent = b"".join(lines)
+        assert sent.startswith(b"GET http://203.0.113.7:80/ HTTP/1.1\r\n")
+        assert b"Connection: close" in sent
+        assert b"keep-alive" not in sent
+        assert b"rebind.evil" not in sent  # the smuggled request never got upstream
+
+
+def test_upstream_proxy_env_parsing(monkeypatch):
+    assert egress_proxy.upstream_proxy() is None
+    monkeypatch.setenv("HTTP_PROXY", "http://192.168.180.254:56560")
+    assert egress_proxy.upstream_proxy() == ("192.168.180.254", 56560, None)
+    monkeypatch.setenv("HTTPS_PROXY", "http://user:p%40ss@10.0.0.1:8080")
+    host, port, auth = egress_proxy.upstream_proxy()
+    assert (host, port) == ("10.0.0.1", 8080)
+    import base64
+    assert base64.b64decode(auth.split(b" ")[-1].strip()) == b"user:p@ss"
+    # scheme-aware selection: http targets prefer HTTP_PROXY
+    assert egress_proxy.upstream_proxy("http") == ("192.168.180.254", 56560, None)
+    monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY", "proxy.corp:3128")
+    assert egress_proxy.upstream_proxy() == ("proxy.corp", 3128, None)
+    # whitespace-only env var means unset, not a proxy named "   "
+    monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY", "   ")
+    monkeypatch.delenv("HTTP_PROXY")
+    monkeypatch.delenv("HTTPS_PROXY")
+    assert egress_proxy.upstream_proxy() is None
+    # non-latin-1 credentials must not raise (encoded as UTF-8)
+    monkeypatch.delenv("CRAWL4AI_UPSTREAM_PROXY")
+    monkeypatch.setenv("HTTPS_PROXY", "http://u:%E5%AF%86%E7%A0%81@10.0.0.1:8080")
+    assert egress_proxy.upstream_proxy()[2] is not None
+    # NO_PROXY routing: suffix and CIDR entries force a direct dial
+    pin = PinnedTarget("https", "site.corp.example", 443, "203.0.113.7")
+    assert egress_proxy._use_upstream(pin) is not None
+    monkeypatch.setenv("NO_PROXY", ".corp.example")
+    assert egress_proxy._use_upstream(pin) is None
+    monkeypatch.setenv("NO_PROXY", "203.0.113.0/24")
+    assert egress_proxy._use_upstream(pin) is None
+
+
 class TestEnforceEgressWiring:
     def test_enforce_egress_sets_proxy(self, monkeypatch):
         import egress_broker
@@ -130,3 +287,565 @@ def test_enforce_egress_sets_proxy(self, monkeypatch):
         egress_broker.enforce_egress(b)
         assert b.proxy_config is not None
         assert b.proxy_config.server == "http://127.0.0.1:9999"
+
+
+@pytest.mark.asyncio
+class TestHostnamePassthrough:
+    """CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES relaxes the pin for listed names only.
+
+    The mode exists for upstreams that front a network whose names do not
+    resolve here, so for those names the upstream must receive the hostname.
+    Everything the allowlist does not name keeps resolve-and-pin, which is what
+    these tests are really guarding.
+    """
+
+    async def test_allowlisted_name_reaches_upstream_unresolved(self, monkeypatch):
+        seen = []
+        corp, corp_port = await _fake_corporate_proxy(seen)
+        monkeypatch.setenv("HTTPS_PROXY", f"http://127.0.0.1:{corp_port}")
+        monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES", ".corp.example")
+
+        def must_not_resolve(url):
+            raise AssertionError(f"resolve_and_pin must not run for {url}")
+        monkeypatch.setattr(egress_proxy, "resolve_and_pin", must_not_resolve)
+
+        proxy = PinningProxy()
+        await proxy.start()
+        try:
+            r, w = await asyncio.open_connection(proxy.bound_host, proxy.bound_port)
+            w.write(b"CONNECT wiki.corp.example:443 HTTP/1.1\r\n\r\n")
+            await w.drain()
+            assert b"200" in await asyncio.wait_for(r.readline(), timeout=5)
+            w.close()
+        finally:
+            await proxy.stop()
+            corp.close()
+        # The hostname is what the upstream must see; it does the lookup.
+        assert seen == [b"CONNECT wiki.corp.example:443 HTTP/1.1\r\n"]
+
+    async def test_name_outside_allowlist_still_pinned(self, monkeypatch):
+        seen = []
+        corp, corp_port = await _fake_corporate_proxy(seen)
+        monkeypatch.setenv("HTTPS_PROXY", f"http://127.0.0.1:{corp_port}")
+        monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES", ".corp.example")
+
+        monkeypatch.setattr(
+            egress_proxy, "resolve_and_pin",
+            lambda url: PinnedTarget("https", "other.example", 443, "203.0.113.9"),
+        )
+
+        proxy = PinningProxy()
+        await proxy.start()
+        try:
+            r, w = await asyncio.open_connection(proxy.bound_host, proxy.bound_port)
+            w.write(b"CONNECT other.example:443 HTTP/1.1\r\n\r\n")
+            await w.drain()
+            assert b"200" in await asyncio.wait_for(r.readline(), timeout=5)
+            w.close()
+        finally:
+            await proxy.stop()
+            corp.close()
+        assert seen == [b"CONNECT 203.0.113.9:443 HTTP/1.1\r\n"]
+
+    async def test_ip_literal_never_bypasses_the_pin(self, monkeypatch):
+        """An address carries no DNS to defer, so passthrough must not apply —
+        otherwise a suffix entry would hand 169.254.169.254 straight through."""
+        seen = []
+        corp, corp_port = await _fake_corporate_proxy(seen)
+        monkeypatch.setenv("HTTPS_PROXY", f"http://127.0.0.1:{corp_port}")
+        monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES", "*,.corp.example")
+
+        def fake_pin(url):
+            raise EgressBlocked()
+        monkeypatch.setattr(egress_proxy, "resolve_and_pin", fake_pin)
+
+        proxy = PinningProxy()
+        await proxy.start()
+        try:
+            r, w = await asyncio.open_connection(proxy.bound_host, proxy.bound_port)
+            w.write(b"CONNECT 169.254.169.254:443 HTTP/1.1\r\n\r\n")
+            await w.drain()
+            assert b"403" in await asyncio.wait_for(r.readline(), timeout=5)
+            w.close()
+        finally:
+            await proxy.stop()
+            corp.close()
+        assert seen == []
+
+    async def test_unset_variable_changes_nothing(self, monkeypatch):
+        seen = []
+        corp, corp_port = await _fake_corporate_proxy(seen)
+        monkeypatch.setenv("HTTPS_PROXY", f"http://127.0.0.1:{corp_port}")
+
+        monkeypatch.setattr(
+            egress_proxy, "resolve_and_pin",
+            lambda url: PinnedTarget("https", "wiki.corp.example", 443, "203.0.113.7"),
+        )
+
+        proxy = PinningProxy()
+        await proxy.start()
+        try:
+            r, w = await asyncio.open_connection(proxy.bound_host, proxy.bound_port)
+            w.write(b"CONNECT wiki.corp.example:443 HTTP/1.1\r\n\r\n")
+            await w.drain()
+            assert b"200" in await asyncio.wait_for(r.readline(), timeout=5)
+            w.close()
+        finally:
+            await proxy.stop()
+            corp.close()
+        assert seen == [b"CONNECT 203.0.113.7:443 HTTP/1.1\r\n"]
+
+
+class TestUrlValidationDelegation:
+    """The entry-point check must honour the same allowlist as the proxy.
+
+    validate_url_destination runs before the browser starts, so a delegated name
+    has to be exempt there too — otherwise the request never reaches the proxy
+    and the suffix list has no effect. These guard that the exemption stays as
+    narrow as the proxy-side one.
+    """
+
+    def _validate(self, url):
+        import importlib
+        import utils
+        importlib.reload(utils)
+        return utils.validate_url_destination(url)
+
+    def test_delegated_suffix_is_allowed_through(self, monkeypatch):
+        monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES", ".corp.example")
+        # Would otherwise be rejected: the name does not resolve here.
+        self._validate("https://wiki.corp.example/page")
+
+    def test_host_outside_the_list_still_blocked(self, monkeypatch):
+        from fastapi import HTTPException
+        monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES", ".corp.example")
+        with pytest.raises(HTTPException):
+            self._validate("http://localhost/")
+
+    def test_ip_literal_still_blocked(self, monkeypatch):
+        from fastapi import HTTPException
+        monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES", "*,.corp.example")
+        with pytest.raises(HTTPException):
+            self._validate("http://169.254.169.254/latest/meta-data/")
+
+    def test_unset_leaves_validation_untouched(self, monkeypatch):
+        from fastapi import HTTPException
+        monkeypatch.delenv("CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES", raising=False)
+        with pytest.raises(HTTPException):
+            self._validate("http://127.0.0.1/")
+
+
+class TestSuffixMatching:
+    """Equivalent spellings of one name must not disagree with the allowlist.
+
+    A suffix list is only as good as the comparison behind it: if "app.internal."
+    and "app.internal" are treated as different names, an operator's list silently
+    misses one of them. These pin the normalisation so a match means the same
+    thing on both sides, and so the near-misses stay misses.
+    """
+
+    def _matches(self, suffixes, host, monkeypatch):
+        import egress_proxy
+        monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES", suffixes)
+        allowed = egress_proxy.passthrough_suffixes()
+        normalized = egress_proxy.normalize_host(host)
+        return bool(normalized) and any(
+            normalized == s or normalized.endswith("." + s) for s in allowed
+        )
+
+    @pytest.mark.parametrize("host", [
+        "app.internal",
+        "sub.app.internal",
+        "APP.INTERNAL",
+        "app.internal.",       # root dot: the same name, written absolutely
+        "APP.INTERNAL.",
+    ])
+    def test_equivalent_spellings_all_match(self, host, monkeypatch):
+        assert self._matches(".internal", host, monkeypatch)
+
+    @pytest.mark.parametrize("host", [
+        "app.internal.attacker.com",   # suffix in the middle, not at the end
+        "fakeinternal",                # no dot boundary
+        "notapp.internal.evil.net",
+        "",
+    ])
+    def test_near_misses_stay_pinned(self, host, monkeypatch):
+        assert not self._matches(".internal", host, monkeypatch)
+
+    @pytest.mark.parametrize("host", ["wiki.corp.example", "wiki.corp.example."])
+    def test_both_layers_read_one_name_identically(self, host, monkeypatch):
+        """A name delegated by the proxy must also be exempt at the entry point.
+
+        The two layers run in different processes and were written apart; if they
+        disagreed, a name would pass one and be rejected by the other and the
+        setting would look broken rather than unsafe. Assert them together.
+        """
+        import importlib
+        import egress_proxy
+        import utils
+        monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES", ".corp.example")
+        monkeypatch.setenv("HTTPS_PROXY", "http://upstream:3128")
+        monkeypatch.delenv("NO_PROXY", raising=False)
+        importlib.reload(utils)
+        assert egress_proxy._passthrough_upstream("https", host, 443) == (
+            ("upstream", 3128, None), "wiki.corp.example")
+        assert utils._delegated_to_upstream(f"https://{host}/page")
+
+    @pytest.mark.parametrize("entry", ["*", "*.internal", "a*b"])
+    def test_wildcards_are_dropped_not_honoured(self, entry, monkeypatch):
+        """A wildcard would delegate every name; that is the one shape to refuse."""
+        import egress_proxy
+        monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES", entry)
+        assert egress_proxy.passthrough_suffixes() == []
+
+    def test_wildcard_does_not_poison_the_rest_of_the_list(self, monkeypatch):
+        import egress_proxy
+        monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES", "*,.corp.example")
+        assert egress_proxy.passthrough_suffixes() == ["corp.example"]
+
+    @pytest.mark.parametrize("host", ["169.254.169.254", "127.0.0.1", "::1"])
+    def test_listing_an_ip_literal_does_not_bypass_the_pin(self, host, monkeypatch):
+        """Even an operator who lists an address gets resolve-and-pin for it.
+
+        An IP literal carries no DNS to defer, so there is nothing passthrough
+        could be for; honouring one would hand 169.254.169.254 to the upstream
+        verbatim and give up the non-global rule for nothing in return. Both
+        layers reject the literal before the suffix list is ever consulted.
+        """
+        import importlib
+        import egress_proxy
+        import utils
+        from fastapi import HTTPException
+        monkeypatch.setenv(
+            "CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES", f".internal,{host}")
+        monkeypatch.setenv("HTTPS_PROXY", "http://upstream:3128")
+        importlib.reload(utils)
+        assert egress_proxy._passthrough_upstream("https", host, 443) is None
+        with pytest.raises(HTTPException):
+            utils.validate_url_destination(f"http://{host}/")
+
+
+class TestPassthroughPorts:
+    """Passthrough gives up the address check, so the port is what still narrows it.
+
+    On the pinned path any port is reachable but only at a global address. A
+    delegated name has no such floor, so without a port policy a listed suffix
+    would expose whatever the upstream can reach on 9200, 5432 or 2375 — not
+    just its web servers. Default to the web ports and let operators widen it.
+    """
+
+    EXPECTED = (("upstream", 3128, None), "wiki.corp.example")
+
+    def _upstream(self, port, monkeypatch, ports_env=None):
+        import egress_proxy
+        monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES", ".corp.example")
+        monkeypatch.setenv("HTTPS_PROXY", "http://upstream:3128")
+        monkeypatch.delenv("NO_PROXY", raising=False)
+        if ports_env is None:
+            monkeypatch.delenv("CRAWL4AI_UPSTREAM_PROXY_DNS_PORTS", raising=False)
+        else:
+            monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_DNS_PORTS", ports_env)
+        return egress_proxy._passthrough_upstream("https", "wiki.corp.example", port)
+
+    @pytest.mark.parametrize("port", [80, 443])
+    def test_web_ports_are_delegated_by_default(self, port, monkeypatch):
+        assert self._upstream(port, monkeypatch) == self.EXPECTED
+
+    @pytest.mark.parametrize("port", [22, 2375, 5432, 6379, 9200, 8080])
+    def test_other_ports_fall_back_to_the_pin(self, port, monkeypatch):
+        """Not delegated, so the caller resolves and pins it like any other target."""
+        assert self._upstream(port, monkeypatch) is None
+
+    def test_operator_can_widen_the_list(self, monkeypatch):
+        assert self._upstream(8443, monkeypatch, "80,443,8443") == self.EXPECTED
+
+    def test_widening_replaces_rather_than_extends(self, monkeypatch):
+        """An explicit list is the whole policy, so 443 is gone unless named."""
+        assert self._upstream(443, monkeypatch, "8443") is None
+
+    @pytest.mark.parametrize("value", ["nope", "0", "70000", "-1", "80x,443x"])
+    def test_a_set_but_unusable_list_closes_rather_than_opens(self, value, monkeypatch):
+        """The operator asked for a policy we could not honour; do not fall back."""
+        assert self._upstream(443, monkeypatch, value) is None
+
+    @pytest.mark.parametrize("value", ["", "   "])
+    def test_blank_means_unset_not_empty(self, value, monkeypatch):
+        """Blank is how an unset variable arrives; it must not disable the mode."""
+        assert self._upstream(443, monkeypatch, value) == self.EXPECTED
+
+    def test_one_bad_entry_does_not_drop_the_good_ones(self, monkeypatch):
+        assert self._upstream(443, monkeypatch, "nope,443") == self.EXPECTED
+
+    def test_entry_point_applies_the_same_port_policy(self, monkeypatch):
+        """Otherwise a URL would pass validation and then be refused by the proxy."""
+        import importlib
+        import utils
+        from fastapi import HTTPException
+        monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES", ".corp.example")
+        monkeypatch.delenv("CRAWL4AI_UPSTREAM_PROXY_DNS_PORTS", raising=False)
+        importlib.reload(utils)
+        assert utils._delegated_to_upstream("https://wiki.corp.example/page")
+        assert not utils._delegated_to_upstream("https://wiki.corp.example:9200/")
+        with pytest.raises(HTTPException):
+            utils.validate_url_destination("https://wiki.corp.example:9200/")
+
+
+@pytest.mark.asyncio
+@pytest.mark.asyncio
+class TestUpstreamOnlySuffixes:
+    """CRAWL4AI_UPSTREAM_PROXY_ONLY_SUFFIXES narrows chaining to listed names.
+
+    Chaining is otherwise all-or-nothing: with an upstream set every target goes
+    through it and NO_PROXY subtracts exceptions. One deployment therefore cannot
+    serve an internal-only proxy and direct public egress at the same time. This
+    allowlist inverts the rule for the names it holds; everything else dials
+    direct and keeps resolve-and-pin, which is what these tests guard.
+    """
+
+    async def test_unset_variable_changes_nothing(self, monkeypatch):
+        seen = []
+        corp, corp_port = await _fake_corporate_proxy(seen)
+        monkeypatch.setenv("HTTPS_PROXY", f"http://127.0.0.1:{corp_port}")
+        monkeypatch.setattr(
+            egress_proxy, "resolve_and_pin",
+            lambda url: PinnedTarget("https", "any.example", 443, "203.0.113.7"),
+        )
+
+        proxy = PinningProxy()
+        await proxy.start()
+        try:
+            r, w = await asyncio.open_connection(proxy.bound_host, proxy.bound_port)
+            w.write(b"CONNECT any.example:443 HTTP/1.1\r\n\r\n")
+            await w.drain()
+            assert b"200" in await asyncio.wait_for(r.readline(), timeout=5)
+            w.close()
+        finally:
+            await proxy.stop()
+            corp.close()
+        # No allowlist: the upstream is still used for everything, as before.
+        assert seen == [b"CONNECT 203.0.113.7:443 HTTP/1.1\r\n"]
+
+    async def test_listed_suffix_still_reaches_the_upstream(self, monkeypatch):
+        seen = []
+        corp, corp_port = await _fake_corporate_proxy(seen)
+        monkeypatch.setenv("HTTPS_PROXY", f"http://127.0.0.1:{corp_port}")
+        monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_ONLY_SUFFIXES", ".corp.example")
+        monkeypatch.setattr(
+            egress_proxy, "resolve_and_pin",
+            lambda url: PinnedTarget("https", "wiki.corp.example", 443, "203.0.113.8"),
+        )
+
+        proxy = PinningProxy()
+        await proxy.start()
+        try:
+            r, w = await asyncio.open_connection(proxy.bound_host, proxy.bound_port)
+            w.write(b"CONNECT wiki.corp.example:443 HTTP/1.1\r\n\r\n")
+            await w.drain()
+            assert b"200" in await asyncio.wait_for(r.readline(), timeout=5)
+            w.close()
+        finally:
+            await proxy.stop()
+            corp.close()
+        # Listed, so chained — and still as the pinned IP, since only the DNS
+        # suffix list may hand a name over unresolved.
+        assert seen == [b"CONNECT 203.0.113.8:443 HTTP/1.1\r\n"]
+
+    async def test_unlisted_host_never_touches_the_upstream(self, monkeypatch):
+        seen = []
+        corp, corp_port = await _fake_corporate_proxy(seen)
+        monkeypatch.setenv("HTTPS_PROXY", f"http://127.0.0.1:{corp_port}")
+        monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_ONLY_SUFFIXES", ".corp.example")
+
+        upstream, upstream_port = await _fake_upstream()
+        monkeypatch.setattr(
+            egress_proxy, "resolve_and_pin",
+            lambda url: PinnedTarget("https", "public.example", 443, "127.0.0.1"),
+        )
+        monkeypatch.setattr(egress_proxy, "_bracket", lambda ip: ip)
+
+        proxy = PinningProxy()
+        await proxy.start()
+        try:
+            r, w = await asyncio.open_connection(proxy.bound_host, proxy.bound_port)
+            w.write(f"CONNECT public.example:{upstream_port} HTTP/1.1\r\n\r\n".encode())
+            await w.drain()
+            assert b"200" in await asyncio.wait_for(r.readline(), timeout=5)
+            w.close()
+        finally:
+            await proxy.stop()
+            upstream.close()
+            corp.close()
+        # Not listed: dialled direct, so the corporate proxy saw nothing at all.
+        assert seen == []
+
+    async def test_match_is_on_the_label_boundary(self, monkeypatch):
+        seen = []
+        corp, corp_port = await _fake_corporate_proxy(seen)
+        monkeypatch.setenv("HTTPS_PROXY", f"http://127.0.0.1:{corp_port}")
+        monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_ONLY_SUFFIXES", ".corp.example")
+
+        upstream, upstream_port = await _fake_upstream()
+        monkeypatch.setattr(
+            egress_proxy, "resolve_and_pin",
+            lambda url: PinnedTarget("https", "notcorp.example", 443, "127.0.0.1"),
+        )
+        monkeypatch.setattr(egress_proxy, "_bracket", lambda ip: ip)
+
+        proxy = PinningProxy()
+        await proxy.start()
+        try:
+            r, w = await asyncio.open_connection(proxy.bound_host, proxy.bound_port)
+            w.write(f"CONNECT notcorp.example:{upstream_port} HTTP/1.1\r\n\r\n".encode())
+            await w.drain()
+            assert b"200" in await asyncio.wait_for(r.readline(), timeout=5)
+            w.close()
+        finally:
+            await proxy.stop()
+            upstream.close()
+            corp.close()
+        # "notcorp.example" is not under "corp.example"; a substring must not match.
+        assert seen == []
+
+    async def test_delegation_requires_both_allowlists(self, monkeypatch):
+        seen = []
+        corp, corp_port = await _fake_corporate_proxy(seen)
+        monkeypatch.setenv("HTTPS_PROXY", f"http://127.0.0.1:{corp_port}")
+        # Delegation is permitted for the name, but the upstream is not.
+        monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES", ".corp.example")
+        monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_ONLY_SUFFIXES", ".other.example")
+
+        upstream, upstream_port = await _fake_upstream()
+        resolved = []
+
+        def pin(url):
+            resolved.append(url)
+            return PinnedTarget("https", "wiki.corp.example", upstream_port, "127.0.0.1")
+        monkeypatch.setattr(egress_proxy, "resolve_and_pin", pin)
+        monkeypatch.setattr(egress_proxy, "_bracket", lambda ip: ip)
+
+        proxy = PinningProxy()
+        await proxy.start()
+        try:
+            r, w = await asyncio.open_connection(proxy.bound_host, proxy.bound_port)
+            w.write(f"CONNECT wiki.corp.example:{upstream_port} HTTP/1.1\r\n\r\n".encode())
+            await w.drain()
+            assert b"200" in await asyncio.wait_for(r.readline(), timeout=5)
+            w.close()
+        finally:
+            await proxy.stop()
+            upstream.close()
+            corp.close()
+        # The name was pinned rather than handed over: one list cannot widen the
+        # other, so an upstream the name may not use never receives it unresolved.
+        assert resolved, "resolve_and_pin must run when the upstream is not allowed"
+        assert seen == []
+
+def test_upstream_only_suffixes_parsing(monkeypatch):
+    monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_ONLY_SUFFIXES", "*")
+    assert egress_proxy.upstream_only_suffixes() == []
+    # An ignored list is an absent list: the pre-existing rule applies, which
+    # chains everything. It never becomes a way to skip the pin.
+    assert egress_proxy._upstream_allows("anything.example") is True
+
+    monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_ONLY_SUFFIXES", ".corp.example,not_a_host!,*.x")
+    assert egress_proxy.upstream_only_suffixes() == ["corp.example"]
+    assert egress_proxy._upstream_allows("wiki.corp.example") is True
+    assert egress_proxy._upstream_allows("elsewhere.example") is False
+    # An IP literal carries no suffix to match, so it is never chained here.
+    assert egress_proxy._upstream_allows("203.0.113.9") is False
+
+
+class TestDelegatedNameOnTheWire:
+    """What the upstream receives must be the name the allowlist authorised.
+
+    Matching on a normalised name while sending the caller's original spelling
+    would mean the allowlist and the connection are about two different strings.
+    These assert the bytes the upstream actually sees, not just the decision.
+    """
+
+    async def test_connect_sends_the_authorised_name(self, monkeypatch):
+        seen = []
+        corp, corp_port = await _fake_corporate_proxy(seen)
+        monkeypatch.setenv("HTTPS_PROXY", f"http://127.0.0.1:{corp_port}")
+        monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES", ".internal")
+        monkeypatch.delenv("NO_PROXY", raising=False)
+
+        def fake_pin(url):
+            raise AssertionError("delegated name must not be resolved locally")
+        monkeypatch.setattr(egress_proxy, "resolve_and_pin", fake_pin)
+
+        proxy = PinningProxy()
+        await proxy.start()
+        try:
+            # Uppercase, a trailing root dot, and a unicode label: three
+            # spellings the allowlist accepts, one name it authorised.
+            for raw in ("WIKI.INTERNAL", "wiki.internal.", "café.internal"):
+                r, w = await asyncio.open_connection(proxy.bound_host, proxy.bound_port)
+                w.write(f"CONNECT {raw}:443 HTTP/1.1\r\n\r\n".encode("latin-1"))
+                await w.drain()
+                assert b"200" in await asyncio.wait_for(r.readline(), timeout=5)
+                w.close()
+        finally:
+            await proxy.stop()
+            corp.close()
+
+        assert seen == [
+            b"CONNECT wiki.internal:443 HTTP/1.1\r\n",
+            b"CONNECT wiki.internal:443 HTTP/1.1\r\n",
+            b"CONNECT xn--caf-dma.internal:443 HTTP/1.1\r\n",
+        ]
+
+    async def test_plain_http_sends_the_authorised_name_and_closes(self, monkeypatch):
+        """Absolute form, Host header, upstream auth and Connection: close.
+
+        A delegated request needs the close as much as a pinned one: the suffix
+        and port were checked for this request only, so a reused connection must
+        not carry a second, unchecked one upstream.
+        """
+        lines = []
+
+        async def handle(reader, writer):
+            req = b""
+            while b"\r\n\r\n" not in req:
+                chunk = await reader.read(4096)
+                if not chunk:
+                    break
+                req += chunk
+            lines.append(req)
+            writer.write(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nhi")
+            await writer.drain()
+            writer.close()
+        corp = await asyncio.start_server(handle, "127.0.0.1", 0)
+        corp_port = corp.sockets[0].getsockname()[1]
+        monkeypatch.setenv("HTTP_PROXY", f"http://user:pw@127.0.0.1:{corp_port}")
+        monkeypatch.setenv("CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES", ".internal")
+        monkeypatch.delenv("NO_PROXY", raising=False)
+
+        def fake_pin(url):
+            raise AssertionError("delegated name must not be resolved locally")
+        monkeypatch.setattr(egress_proxy, "resolve_and_pin", fake_pin)
+
+        proxy = PinningProxy()
+        await proxy.start()
+        try:
+            r, w = await asyncio.open_connection(proxy.bound_host, proxy.bound_port)
+            w.write("GET http://WIKI.INTERNAL./p HTTP/1.1\r\n"
+                    "Host: WIKI.INTERNAL.\r\nConnection: keep-alive\r\n\r\n".encode("latin-1"))
+            await w.drain()
+            assert b"200" in await asyncio.wait_for(r.read(200), timeout=5)
+            # Try to smuggle an unchecked request onto the same connection.
+            w.write(b"GET http://evil.example/ HTTP/1.1\r\nHost: evil.example\r\n\r\n")
+            await w.drain()
+            assert await asyncio.wait_for(r.read(200), timeout=5) == b""
+            w.close()
+        finally:
+            await proxy.stop()
+            corp.close()
+
+        sent = b"".join(lines)
+        assert sent.startswith(b"GET http://wiki.internal:80/p HTTP/1.1\r\n")
+        assert b"Host: wiki.internal\r\n" in sent
+        assert b"Proxy-Authorization: Basic " in sent
+        assert b"Connection: close" in sent
+        assert b"keep-alive" not in sent
+        assert b"evil.example" not in sent
diff --git a/deploy/docker/utils.py b/deploy/docker/utils.py
index 8f8ecb180..e3fad4785 100644
--- a/deploy/docker/utils.py
+++ b/deploy/docker/utils.py
@@ -351,14 +351,57 @@ def get_llm_base_url(config: Dict, provider: Optional[str] = None) -> Optional[s
 ALLOW_INTERNAL_URLS = os.environ.get("CRAWL4AI_ALLOW_INTERNAL_URLS", "false").lower() == "true"
 
 
+def _delegated_to_upstream(url: str) -> bool:
+    """True when this host was delegated to the upstream proxy by the operator.
+
+    CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES lists names an upstream resolves on our
+    behalf (see egress_proxy). Such names are expected not to resolve here, so
+    checking them locally rejects precisely the targets the operator configured.
+    Narrow by construction: an empty list disables it, only listed suffixes
+    match, only the allowed ports qualify, and an IP literal never does — there
+    is no name to delegate, so link-local and metadata addresses stay blocked
+    whatever is listed.
+
+    The parsing and matching are imported from egress_proxy rather than repeated
+    here, so this check and the proxy cannot drift into disagreeing about which
+    names are delegated — a disagreement would be a hole, not a mismatch.
+    """
+    from egress_proxy import normalize_host, passthrough_ports, passthrough_suffixes
+
+    suffixes = passthrough_suffixes()
+    if not suffixes:
+        return False
+    try:
+        parsed = urlparse(str(url))
+        raw_host = parsed.hostname or ""
+        port = parsed.port or (443 if parsed.scheme == "https" else 80)
+    except ValueError:
+        # An unparseable port names no delegated target; fall back to the check.
+        return False
+    if port not in passthrough_ports():
+        return False
+    try:
+        ipaddress.ip_address(raw_host.strip().rstrip("."))
+        return False
+    except ValueError:
+        pass
+    host = normalize_host(raw_host)
+    if not host:
+        return False
+    return any(host == s or host.endswith("." + s) for s in suffixes)
+
+
 def validate_url_destination(url: str) -> None:
     """Block crawl URLs targeting internal/private networks (SSRF protection).
     Skipped when CRAWL4AI_ALLOW_INTERNAL_URLS=true.
+    Skipped for hosts delegated via CRAWL4AI_UPSTREAM_PROXY_DNS_SUFFIXES.
     Skipped for raw: URLs (inline HTML, no network fetch)."""
     if ALLOW_INTERNAL_URLS:
         return
     if str(url).startswith(("raw:", "raw://")):
         return
+    if _delegated_to_upstream(url):
+        return
     try:
         validate_webhook_url(url)
     except ValueError as e: