Docker: Chain egress proxy through upstream HTTP(S)_PROXY

deploy/docker/README.md

@@ -116,6 +116,17 @@ 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. Proxies that refuse CONNECT-to-an-IP, or
+    containers with no DNS at all, are not yet supported.
+
 #### 4. Stopping the Container
 
 ```bash

deploy/docker/egress_proxy.py

@@ -13,13 +13,21 @@
 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.
 """
 
 from __future__ import annotations
 
 import asyncio
+import base64
+import ipaddress
 import logging
-from urllib.parse import urlsplit
+import os
+from urllib.parse import unquote, urlsplit
 
 from egress_broker import EgressBlocked, resolve_and_pin
 
@@ -31,6 +39,63 @@
 _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 _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
+    return up
+
+
+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 +117,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:
@@ -101,9 +172,7 @@ async def _handle_connect(self, target, client_reader, client_writer):
         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
-            )
+            up_reader, up_writer = await self._dial(pin, int(port_s))
         except Exception:
             await self._reply(client_writer, _BLOCKED)
             return
@@ -129,15 +198,31 @@ async def _handle_absolute(self, method, target, request_line, client_reader, cl
         path = sp.path or "/"
         if sp.query:
             path += "?" + sp.query
+        upstream = _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")
+        # 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:
+            out = f"{method} http://{_bracket(pin.ip)}:{port}{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.
+            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"
         out += b"Host: " + sp.hostname.encode("latin-1")
         if sp.port:
             out += f":{sp.port}".encode("latin-1")
@@ -147,6 +232,39 @@ 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(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:

deploy/docker/tests/test_security_egress_proxy.py

@@ -20,6 +20,18 @@
 
 pytestmark = pytest.mark.posture
 
+_PROXY_ENV = (
+    "CRAWL4AI_UPSTREAM_PROXY", "HTTP_PROXY", "http_proxy",
+    "HTTPS_PROXY", "https_proxy", "NO_PROXY", "no_proxy",
+)
+
+
+@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 +133,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