diff --git a/deploy/docker/README.md b/deploy/docker/README.md index 7ce0dcd00..a57c857b9 100644 --- a/deploy/docker/README.md +++ b/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 diff --git a/deploy/docker/egress_proxy.py b/deploy/docker/egress_proxy.py index 4cdb56bd1..1aa1bcc2f 100644 --- a/deploy/docker/egress_proxy.py +++ b/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: diff --git a/deploy/docker/tests/test_security_egress_proxy.py b/deploy/docker/tests/test_security_egress_proxy.py index 03ebae1ea..f53767f35 100644 --- a/deploy/docker/tests/test_security_egress_proxy.py +++ b/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