From 2a9e7255a7c9bb50b8d0bb5b5b5028095f2ad94d Mon Sep 17 00:00:00 2001 From: fuduoduo <3253787109@qq.com> Date: Sun, 8 Nov 2026 22:10:10 +0800 Subject: [PATCH] fix(channels): keep Gateway calls off the environment proxy (#7877) * fix(channels): keep Gateway calls off the environment proxy * fix(channels): keep the aio_sandbox policy import lazy Importing deerflow.community.aio_sandbox.backend at module scope executes the package __init__, which eagerly imports AioSandbox, AioSandboxProvider and both backends. That makes every ChannelManager import load the entire sandbox provider stack even when a different sandbox is configured. Mirror app/gateway/health.py::_make_provisioner_client: move the import into a small _make_gateway_client helper and route both Gateway call sites through it, so the URL policy stays lazy. * fix(channels): keep the run client off the environment proxy langgraph_sdk.get_client() builds httpx.AsyncHTTPTransport(retries=5) on an httpx.AsyncClient that leaves trust_env at its default True and exposes no transport/trust_env override, so ChannelManager._get_client() still sent runs.wait submissions to a loopback Gateway through HTTP_PROXY. That is the hop #7876 reports: the control-plane calls are fixed, the run submission is not. Route the SDK client through _make_langgraph_client: loopback and private Gateways get an explicit trust_env=False httpx.AsyncClient carrying the SDK's own retries, timeout and User-Agent and handed to LangGraphClient, while external Gateways keep the stock get_client() path. Also move the URL policy out of the aio_sandbox package into deerflow.utils.http_policy (control_plane_trust_env). Keeping the import lazy did not help on its own: aio_sandbox/__init__ eagerly imports the provider stack, so importing aio_sandbox.backend still loaded all of it. The neutral module has no httpx or sandbox dependency, so manager.py and gateway/health.py import it at module scope, and aio_sandbox.backend re-exports it as sandbox_http_trust_env for the existing sandbox call sites. * fix(channels): keep AGENTS.md under budget and align legacy CSRF test with new client * style(sandbox): remove unused urlparse import --------- Co-authored-by: jack12178 Co-authored-by: Willem Jiang --- backend/app/channels/manager.py | 66 ++++++- backend/app/gateway/health.py | 6 +- .../deerflow/community/aio_sandbox/backend.py | 32 +--- .../packages/harness/deerflow/utils/AGENTS.md | 25 +++ .../harness/deerflow/utils/http_policy.py | 40 ++++ backend/tests/test_channels.py | 172 +++++++++++++++++- 6 files changed, 303 insertions(+), 38 deletions(-) create mode 100644 backend/packages/harness/deerflow/utils/http_policy.py diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index f8656e0d53..f0e139a6b8 100644 --- a/backend/app/channels/manager.py +++ b/backend/app/channels/manager.py @@ -69,6 +69,7 @@ from deerflow.skills.storage.skill_storage import SkillStorage from deerflow.trace_context import ensure_trace_context from deerflow.uploads.manager import apply_upload_sandbox_permits from deerflow.utils.file_io import await_drained +from deerflow.utils.http_policy import control_plane_trust_env from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY logger = logging.getLogger(__name__) @@ -301,6 +302,59 @@ _WECOM_MEDIA_COS_HOST_RE = re.compile(r"^ww-aibot-img-(?P\d+)\.cos\.[a-z0 _METADATA_DROP_KEYS = frozenset({"raw_message", "ref_msg"}) +def _make_gateway_client(gateway_url: str) -> httpx.AsyncClient: + """Build the short-lived client for Gateway control-plane calls. + + A loopback Gateway must bypass ``HTTP_PROXY``, while an external one keeps + the environment proxy. The policy itself lives in + :mod:`deerflow.utils.http_policy` so this module never imports the sandbox + provider stack. + """ + return httpx.AsyncClient(trust_env=control_plane_trust_env(gateway_url)) + + +# Mirrors the client ``langgraph_sdk.get_client()`` builds, so the loopback path +# below differs from it only in ``trust_env``. +_LANGGRAPH_SDK_RETRIES = 5 +_LANGGRAPH_SDK_TIMEOUT = httpx.Timeout(connect=5, read=300, write=300, pool=5) + + +def _make_langgraph_client(base_url: str, headers: Mapping[str, str]) -> Any: + """Build the ``langgraph_sdk`` async client used for channel run submission. + + ``get_client()`` always builds ``httpx.AsyncHTTPTransport(retries=5)`` on an + ``httpx.AsyncClient`` that keeps ``trust_env`` at its default ``True`` and + exposes neither ``transport`` nor ``trust_env``, so a loopback Gateway would + still be sent through ``HTTP_PROXY``: control-plane calls are fixed but + ``runs.wait`` keeps hitting the proxy-generated error of #7876. Loopback and + private Gateways therefore get an explicit ``trust_env=False`` client handed + to the SDK; external Gateways keep the stock ``get_client()`` path. + """ + from langgraph_sdk import __version__ as sdk_version + from langgraph_sdk import get_client + + if control_plane_trust_env(base_url): + return get_client(url=base_url, headers=dict(headers)) + + try: # langgraph-sdk keeps the async client in a private package since 0.3 + from langgraph_sdk._async import LangGraphClient + except ImportError: # older releases expose it directly on ``langgraph_sdk.client`` + from langgraph_sdk.client import LangGraphClient + + # ``get_client`` also folds an ``LANGGRAPH_API_KEY`` env var into the SDK + # headers; a control-plane Gateway authenticates with the internal auth + # headers above, so only those are forwarded here. + return LangGraphClient( + httpx.AsyncClient( + base_url=base_url, + transport=httpx.AsyncHTTPTransport(retries=_LANGGRAPH_SDK_RETRIES), + timeout=_LANGGRAPH_SDK_TIMEOUT, + headers={"User-Agent": f"langgraph-sdk-py/{sdk_version}", **headers}, + trust_env=False, + ) + ) + + def _slim_metadata(meta: dict[str, Any]) -> dict[str, Any]: """Return a shallow copy of *meta* with known-large keys removed.""" return {k: v for k, v in meta.items() if k not in _METADATA_DROP_KEYS} @@ -2318,11 +2372,9 @@ class ChannelManager: def _get_client(self): """Return the ``langgraph_sdk`` async client, creating it on first use.""" if self._client is None: - from langgraph_sdk import get_client - - self._client = get_client( - url=self._langgraph_url, - headers={**create_internal_auth_headers(), **self._csrf_headers()}, + self._client = _make_langgraph_client( + self._langgraph_url, + {**create_internal_auth_headers(), **self._csrf_headers()}, ) return self._client @@ -4056,7 +4108,7 @@ class ChannelManager: headers: dict[str, str], json: dict[str, Any] | None = None, ) -> dict[str, Any]: - async with httpx.AsyncClient() as http: + async with _make_gateway_client(self._gateway_url) as http: request = getattr(http, method.lower()) kwargs: dict[str, Any] = {"timeout": 10, "headers": {**headers, **self._csrf_headers()}} if json is not None: @@ -4152,7 +4204,7 @@ class ChannelManager: replies should catch and degrade (see ``_fetch_gateway``). """ headers = _owner_headers(msg) if msg is not None else None - async with httpx.AsyncClient() as http: + async with _make_gateway_client(self._gateway_url) as http: resp = await http.get( f"{self._gateway_url}{path}", timeout=10, diff --git a/backend/app/gateway/health.py b/backend/app/gateway/health.py index 729e0340ef..7667a90275 100644 --- a/backend/app/gateway/health.py +++ b/backend/app/gateway/health.py @@ -51,6 +51,7 @@ from sqlalchemy import text from deerflow.persistence.engine import get_engine from deerflow.utils.file_io import await_drained +from deerflow.utils.http_policy import control_plane_trust_env if TYPE_CHECKING: from deerflow.config.app_config import AppConfig @@ -313,10 +314,7 @@ def _make_provisioner_client(provisioner_url: str) -> httpx.AsyncClient: ``NO_PROXY``, so without this a private provisioner behind a proxy-only egress would be reported unreachable on every probe. """ - # Lazy: importing the aio_sandbox package pulls in the whole provider stack. - from deerflow.community.aio_sandbox.backend import sandbox_http_trust_env - - return httpx.AsyncClient(timeout=_PROBE_TIMEOUT_SECONDS, trust_env=sandbox_http_trust_env(provisioner_url)) + return httpx.AsyncClient(timeout=_PROBE_TIMEOUT_SECONDS, trust_env=control_plane_trust_env(provisioner_url)) async def _probe_provisioner(provisioner_url: str | None) -> str: diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/backend.py b/backend/packages/harness/deerflow/community/aio_sandbox/backend.py index db56b332b1..db7cb3f6b4 100644 --- a/backend/packages/harness/deerflow/community/aio_sandbox/backend.py +++ b/backend/packages/harness/deerflow/community/aio_sandbox/backend.py @@ -3,44 +3,26 @@ from __future__ import annotations import asyncio -import ipaddress import logging import time from abc import ABC, abstractmethod from collections.abc import Mapping from typing import Any -from urllib.parse import urlparse import httpx import requests +from deerflow.utils.http_policy import control_plane_trust_env + from .sandbox_info import SandboxInfo logger = logging.getLogger(__name__) - -def sandbox_http_trust_env(sandbox_url: str) -> bool: - """Whether HTTP clients for *sandbox_url* should inherit proxy settings. - - Local Docker, DooD, and Kubernetes sandbox endpoints are control-plane - connections, not internet traffic. Sending them through ``HTTP_PROXY`` can - produce a misleading proxy-generated 502 even though the sandbox container - is healthy (#3441). External fully-qualified hosts retain normal environment - proxy behavior. - """ - try: - hostname = (urlparse(sandbox_url).hostname or "").rstrip(".").lower() - except ValueError: - return True - if not hostname: - return True - if hostname == "localhost" or hostname.endswith(".localhost") or hostname.endswith(".docker.internal") or hostname.endswith(".containers.internal"): - return False - try: - address = ipaddress.ip_address(hostname) - except ValueError: - return "." in hostname - return not (address.is_loopback or address.is_private or address.is_link_local) +# The policy lives in ``deerflow.utils.http_policy`` so Gateway-side callers can +# apply it without importing this package (whose ``__init__`` loads the whole +# sandbox provider stack). Kept as a module-level name for the existing sandbox +# call sites and for tests that patch it here. +sandbox_http_trust_env = control_plane_trust_env # The readiness deadline the local-container provider paths (sync and async) diff --git a/backend/packages/harness/deerflow/utils/AGENTS.md b/backend/packages/harness/deerflow/utils/AGENTS.md index 2bad017b35..ffa042d03a 100644 --- a/backend/packages/harness/deerflow/utils/AGENTS.md +++ b/backend/packages/harness/deerflow/utils/AGENTS.md @@ -76,6 +76,20 @@ when all remaining ports are reserved or occupied, without probing invalid TCP ports. Keep the last valid port allocatable and preserve `max_range`. Regression coverage lives in `tests/test_port_allocator_bounds.py`. +### Control-Plane Proxy Policy + +`http_policy.py::control_plane_trust_env` is the single answer to "should this +client inherit `HTTP_PROXY`?" for DeerFlow's own hops. Loopback, private, +link-local, `.localhost`, Docker/Podman-internal and cluster-local hostnames are +control-plane traffic and must bypass the environment proxy; external FQDNs and +public IPs keep it. The module stays dependency-free (no `httpx`, no sandbox +imports) precisely so `app/channels/manager.py` and `app/gateway/health.py` can +import it at module scope without loading the sandbox provider stack. +`community/aio_sandbox/backend.py` re-exports it as `sandbox_http_trust_env` for +existing sandbox call sites. Policy parity with the original sandbox-local +implementation is covered by `tests/test_aio_sandbox_readiness.py`; the +import-graph guard lives in `tests/test_channels.py`. + ### Goal Objective Validation `goal_objective.py` owns the dependency-free normalized 4000-character objective @@ -238,3 +252,14 @@ retries; the helper remains responsible for error classification and cadence. `parse_retry_after_date_ms(raw, now=...)` owns HTTP-date parsing for Claude, Codex and LLM error middleware. Interpret timezone-less asctime as UTC and apply the shared delay bound. Callers retain numeric/header precedence and local backoff. + +### Control-plane HTTP proxy policy + +`http_policy.control_plane_trust_env(url)` decides whether a client inherits +`HTTP_PROXY`: loopback, private and link-local hosts are control-plane traffic +and bypass it; external hosts keep the environment proxy. `app.channels.manager` +uses it for both `_make_gateway_client` and `_make_langgraph_client`. +`langgraph_sdk.get_client()` hard-codes `trust_env=True` and exposes no transport +override, so the `runs.wait` client for local Gateways is assembled on our own +`httpx.AsyncClient` with `trust_env=False`; otherwise #7876 keeps reproducing. +Importing `ChannelManager` must not pull in `deerflow.community.aio_sandbox`. diff --git a/backend/packages/harness/deerflow/utils/http_policy.py b/backend/packages/harness/deerflow/utils/http_policy.py new file mode 100644 index 0000000000..149559728b --- /dev/null +++ b/backend/packages/harness/deerflow/utils/http_policy.py @@ -0,0 +1,40 @@ +"""Proxy policy shared by every control-plane HTTP client. + +DeerFlow reaches its own Gateway, provisioner and sandboxes over the deployment +network, not the public internet. When operators export ``HTTP_PROXY`` for +outbound model/tool traffic, those hops must not be sent through the proxy: the +proxy answers with its own 502/504 while the target is perfectly healthy +(#3441, #7876). + +This module is deliberately dependency-free (no ``httpx``, no sandbox imports) +so that ``app.channels.manager``, ``app.gateway.health`` and the sandbox +backends can all import the policy without dragging in the sandbox provider +stack. +""" + +from __future__ import annotations + +import ipaddress +from urllib.parse import urlparse + + +def control_plane_trust_env(url: str) -> bool: + """Whether HTTP clients for *url* should inherit environment proxy settings. + + Local Docker, DooD, Kubernetes and co-located Gateway endpoints are + control-plane connections, not internet traffic. External fully-qualified + hosts retain normal environment proxy behavior. + """ + try: + hostname = (urlparse(url).hostname or "").rstrip(".").lower() + except ValueError: + return True + if not hostname: + return True + if hostname == "localhost" or hostname.endswith(".localhost") or hostname.endswith(".docker.internal") or hostname.endswith(".containers.internal"): + return False + try: + address = ipaddress.ip_address(hostname) + except ValueError: + return "." in hostname + return not (address.is_loopback or address.is_private or address.is_link_local) diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index d25bb77817..179952a084 100644 --- a/backend/tests/test_channels.py +++ b/backend/tests/test_channels.py @@ -5,6 +5,8 @@ from __future__ import annotations import asyncio import json import logging +import subprocess +import sys import tempfile import threading from concurrent.futures import Future @@ -702,7 +704,9 @@ class TestChannelManager: bus = MessageBus() store = JsonChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") - manager = ChannelManager(bus=bus, store=store, langgraph_url="http://localhost:8001") + # An external Gateway keeps langgraph_sdk.get_client(); loopback hosts + # build a trust_env=False client and are covered by the run-client tests. + manager = ChannelManager(bus=bus, store=store, langgraph_url="http://gateway.example.com/api") with patch("langgraph_sdk.get_client") as get_client: get_client.return_value = object() @@ -711,7 +715,7 @@ class TestChannelManager: get_client.assert_called_once() kwargs = get_client.call_args.kwargs - assert kwargs["url"] == "http://localhost:8001" + assert kwargs["url"] == "http://gateway.example.com/api" headers = kwargs["headers"] csrf_token = headers["X-CSRF-Token"] assert csrf_token @@ -14425,3 +14429,167 @@ class TestHumanInputResponseTranslation: message = _human_input_message_with_channel_response("hello", metadata={}) assert message["content"] == "hello" assert "additional_kwargs" not in message + + +@pytest.mark.parametrize( + ("gateway_url", "expected_trust_env"), + [ + ("http://localhost:8001", False), + ("http://127.0.0.1:8001", False), + ("http://gateway.example.com", True), + ], +) +@pytest.mark.asyncio +async def test_gateway_clients_do_not_proxy_local_gateway( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + gateway_url: str, + expected_trust_env: bool, +) -> None: + """Gateway calls are control-plane traffic and must bypass ``HTTP_PROXY``. + + The Gateway runs next to the channel manager (``http://localhost:8001`` by + default). Inheriting the environment proxy for that hop produces a + proxy-generated error even though the Gateway is healthy -- the same + failure ``sandbox_http_trust_env`` already guards against for sandbox + control-plane calls. + """ + from app.channels import manager as manager_module + from app.channels.manager import ChannelManager + + seen_kwargs: list[dict[str, object]] = [] + + class _JsonResponse: + def raise_for_status(self) -> None: + return None + + def json(self) -> dict[str, object]: + return {"ok": True} + + class RecordingClient: + def __init__(self, *args: object, **kwargs: object) -> None: + seen_kwargs.append(kwargs) + + async def __aenter__(self) -> RecordingClient: + return self + + async def __aexit__(self, *exc_info: object) -> None: + return None + + async def get(self, url: str, **kwargs: object) -> _JsonResponse: + return _JsonResponse() + + async def put(self, url: str, **kwargs: object) -> _JsonResponse: + return _JsonResponse() + + monkeypatch.setattr(manager_module.httpx, "AsyncClient", RecordingClient) + + manager = ChannelManager( + bus=MessageBus(), + store=JsonChannelStore(path=tmp_path / "store.json"), + gateway_url=gateway_url, + ) + + await manager._fetch_gateway_json("/api/models") + await manager._goal_request("put", "thread-1", headers={}, json={"objective": "x"}) + + assert len(seen_kwargs) == 2, "both Gateway call sites should build a client" + assert all(kwargs.get("trust_env") is expected_trust_env for kwargs in seen_kwargs) + + +@pytest.mark.parametrize("langgraph_url", ["http://localhost:8001/api", "http://127.0.0.1:8001/api"]) +def test_run_client_bypasses_env_proxy_for_local_gateway( + monkeypatch: pytest.MonkeyPatch, + langgraph_url: str, +) -> None: + """``runs.wait`` submissions are control-plane traffic too. + + ``langgraph_sdk.get_client()`` builds its own ``httpx.AsyncClient`` with + ``trust_env`` left at the default ``True`` and exposes no override, so a + loopback Gateway was still reached through ``HTTP_PROXY`` -- the exact hop + #7876 reports, and one the Gateway control-plane clients do not cover. + """ + from app.channels import manager as manager_module + + seen_kwargs: list[dict[str, object]] = [] + + class RecordingClient: + def __init__(self, *args: object, **kwargs: object) -> None: + seen_kwargs.append(kwargs) + + monkeypatch.setattr(manager_module.httpx, "AsyncClient", RecordingClient) + + client = manager_module._make_langgraph_client(langgraph_url, {"X-Internal": "1"}) + + assert len(seen_kwargs) == 1, "the SDK client must wrap the client we build" + client_kwargs = seen_kwargs[0] + assert client_kwargs["trust_env"] is False + assert client_kwargs["base_url"] == langgraph_url + headers = client_kwargs["headers"] + assert isinstance(headers, dict) + assert headers["X-Internal"] == "1" + assert headers["User-Agent"].startswith("langgraph-sdk-py/") + assert isinstance(client.http.client, RecordingClient) + + +def test_run_client_keeps_stock_sdk_client_for_external_gateway(monkeypatch: pytest.MonkeyPatch) -> None: + """An external Gateway keeps the SDK's own client, proxy settings included.""" + import langgraph_sdk + + from app.channels import manager as manager_module + + calls: list[dict[str, object]] = [] + sentinel = object() + + def fake_get_client(**kwargs: object) -> object: + calls.append(kwargs) + return sentinel + + monkeypatch.setattr(langgraph_sdk, "get_client", fake_get_client) + + client = manager_module._make_langgraph_client("http://gateway.example.com/api", {"X-Internal": "1"}) + + assert client is sentinel + assert calls == [{"url": "http://gateway.example.com/api", "headers": {"X-Internal": "1"}}] + + +def test_channel_manager_builds_run_client_with_internal_auth_headers( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The run client carries the same internal auth and CSRF pair as control calls.""" + from app.channels import manager as manager_module + from app.channels.manager import ChannelManager + + calls: list[tuple[str, dict[str, str]]] = [] + + def fake_make_langgraph_client(base_url: str, headers: dict[str, str]) -> object: + calls.append((base_url, dict(headers))) + return object() + + monkeypatch.setattr(manager_module, "_make_langgraph_client", fake_make_langgraph_client) + + manager = ChannelManager( + bus=MessageBus(), + store=JsonChannelStore(path=tmp_path / "store.json"), + gateway_url="http://localhost:8001", + ) + manager._get_client() + + assert len(calls) == 1 + base_url, headers = calls[0] + assert base_url == manager_module.DEFAULT_LANGGRAPH_URL + assert manager_module.CSRF_HEADER_NAME in headers + + +def test_importing_channel_manager_loads_no_sandbox_provider_stack() -> None: + """The proxy policy must not drag the sandbox package into every Gateway boot.""" + code = "import sys\nimport app.channels.manager\nprint(','.join(sorted(name for name in sys.modules if name.startswith('deerflow.community.aio_sandbox'))))\n" + result = subprocess.run( + [sys.executable, "-c", code], + cwd=Path(__file__).resolve().parents[1], + capture_output=True, + text=True, + check=True, + ) + assert result.stdout.strip() == "", "ChannelManager import pulled in the sandbox provider stack"