From ddf44b8cf19195c6c85678bfc86bbfe9f10339ee Mon Sep 17 00:00:00 2001 From: evanlowe <62918515+evanlowe@users.noreply.github.com> Date: Tue, 22 Sep 2026 10:43:22 +0800 Subject: [PATCH] fix(cloud): align CloudApp proxy selection with VeFaaS SDK --- tests/cloud/test_cloud_app_proxy.py | 182 ++++++++++++++++++++++++++++ veadk/cloud/cloud_app.py | 14 ++- 2 files changed, 195 insertions(+), 1 deletion(-) create mode 100644 tests/cloud/test_cloud_app_proxy.py diff --git a/tests/cloud/test_cloud_app_proxy.py b/tests/cloud/test_cloud_app_proxy.py new file mode 100644 index 000000000..4412128e8 --- /dev/null +++ b/tests/cloud/test_cloud_app_proxy.py @@ -0,0 +1,182 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import threading +from collections.abc import Iterator +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from unittest.mock import Mock, patch + +import httpx +import pytest + +from veadk.cloud.cloud_agent_engine import CloudAgentEngine +from veadk.cloud.cloud_app import CloudApp + + +@pytest.fixture(autouse=True) +def clean_network_environment(monkeypatch: pytest.MonkeyPatch) -> None: + for key in ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY"): + monkeypatch.delenv(key, raising=False) + monkeypatch.delenv(key.lower(), raising=False) + for key in ("SSL_CERT_FILE", "SSL_CERT_DIR"): + monkeypatch.delenv(key, raising=False) + + +@pytest.fixture +def http_server() -> Iterator[tuple[str, list[tuple[str, str]]]]: + requests: list[tuple[str, str]] = [] + + class Handler(BaseHTTPRequestHandler): + def do_GET(self) -> None: + requests.append((self.command, self.path)) + self.send_response(200) + self.send_header("Content-Length", "2") + self.end_headers() + self.wfile.write(b"ok") + + def do_CONNECT(self) -> None: + requests.append((self.command, self.path)) + # Record the selected HTTPS proxy without contacting the destination + self.send_error(502, "Test proxy") + + def log_message(self, *_args: object) -> None: + pass + + with ThreadingHTTPServer(("127.0.0.1", 0), Handler) as server: + thread = threading.Thread( + target=lambda: server.serve_forever(poll_interval=0.01), daemon=True + ) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_port}", requests + finally: + server.shutdown() + thread.join(timeout=2) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "unused_proxy", ["ALL_PROXY", "all_proxy", "HTTPS_PROXY", "https_proxy"] +) +async def test_cloud_app_connects_directly_with_unrelated_invalid_proxy( + monkeypatch: pytest.MonkeyPatch, + http_server: tuple[str, list[tuple[str, str]]], + unused_proxy: str, +) -> None: + endpoint, requests = http_server + monkeypatch.setenv("HTTP_PROXY", "") + monkeypatch.setenv("http_proxy", "") + monkeypatch.setenv(unused_proxy, "socks://127.0.0.1:7897") + + app = CloudApp(vefaas_endpoint=endpoint) + async with app.httpx_client as client: + response = await client.get(f"{endpoint}/health", timeout=2) + + assert response.text == "ok" + assert requests == [("GET", "/health")] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("scheme", ["http", "https"]) +@pytest.mark.parametrize("proxy_case", ["uppercase", "lowercase", "empty_uppercase"]) +async def test_cloud_app_routes_through_sdk_selected_proxy( + monkeypatch: pytest.MonkeyPatch, + http_server: tuple[str, list[tuple[str, str]]], + scheme: str, + proxy_case: str, +) -> None: + proxy, requests = http_server + upper_key = f"{scheme.upper()}_PROXY" + lower_key = f"{scheme}_proxy" + invalid_proxy = "socks://127.0.0.1:7897" + monkeypatch.setenv("ALL_PROXY", invalid_proxy) + monkeypatch.setenv("all_proxy", invalid_proxy) + # The VeFaaS SDK does not apply NO_PROXY to its selected proxy + monkeypatch.setenv("NO_PROXY", "studio.example") + monkeypatch.setenv( + "HTTPS_PROXY" if scheme == "http" else "HTTP_PROXY", invalid_proxy + ) + if proxy_case == "uppercase": + monkeypatch.setenv(upper_key, proxy) + monkeypatch.setenv(lower_key, invalid_proxy) + else: + monkeypatch.setenv(lower_key, proxy) + if proxy_case == "empty_uppercase": + monkeypatch.setenv(upper_key, "") + + endpoint = f"{scheme}://studio.example/health" + app = CloudApp(vefaas_endpoint=endpoint) + async with app.httpx_client as client: + if scheme == "http": + response = await client.get(endpoint, timeout=2) + assert response.text == "ok" + else: + with pytest.raises(httpx.ProxyError, match="502 Test proxy"): + await client.get(endpoint, timeout=2) + + assert requests == ( + [("GET", endpoint)] if scheme == "http" else [("CONNECT", "studio.example:443")] + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("provider", ["volcengine", "byteplus"]) +async def test_cloud_deploy_returns_app_with_invalid_all_proxy( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, provider: str +) -> None: + monkeypatch.setenv("ALL_PROXY", "socks://127.0.0.1:7897") + endpoint = f"https://studio.{provider}.example" + service = Mock() + service.deploy.return_value = (endpoint, "application-id", "function-id") + service.get_application_route.return_value = ("gateway", "service", "route") + with ( + patch("veadk.cloud.cloud_agent_engine.VeFaaS", return_value=service), + patch("veadk.cloud.cloud_agent_engine.APIGateway"), + patch("veadk.cloud.cloud_agent_engine.IdentityClient"), + ): + engine = CloudAgentEngine( + volcengine_access_key="test_access_key", + volcengine_secret_key="test_secret_key", + provider=provider, + ) + app = engine.deploy("studio-app", str(tmp_path)) + + try: + assert app.vefaas_endpoint == endpoint + assert app.vefaas_application_id == "application-id" + service.deploy.assert_called_once() + finally: + await app.httpx_client.aclose() + + +@pytest.mark.parametrize("scheme", ["http", "https"]) +def test_cloud_app_rejects_invalid_selected_proxy( + monkeypatch: pytest.MonkeyPatch, scheme: str +) -> None: + monkeypatch.setenv(f"{scheme.upper()}_PROXY", "socks://127.0.0.1:7897") + + with pytest.raises(ValueError, match="Unknown scheme for proxy URL"): + CloudApp(vefaas_endpoint=f"{scheme}://studio.example") + + +def test_cloud_app_preserves_custom_certificate_configuration( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + certificate_file = tmp_path / "missing-ca.pem" + monkeypatch.setenv("SSL_CERT_FILE", str(certificate_file)) + + # A missing CA file must still fail, rather than silently using a default CA + with pytest.raises(FileNotFoundError): + CloudApp(vefaas_endpoint="https://studio.example") diff --git a/veadk/cloud/cloud_app.py b/veadk/cloud/cloud_app.py index 9c7a746e8..b308d6b25 100644 --- a/veadk/cloud/cloud_app.py +++ b/veadk/cloud/cloud_app.py @@ -13,6 +13,7 @@ # limitations under the License. import json +import os import time from typing import Any from uuid import uuid4 @@ -46,6 +47,9 @@ class CloudApp: Note: At least one of name, endpoint, or ID must be provided during init. Agent card mode fetches card from the endpoint's public path. + Proxies follow the VeFaaS SDK: use the endpoint's HTTP(S) proxy variable, + preferring uppercase over lowercase. ALL_PROXY, NO_PROXY, and system + proxy settings are not used. Examples: ```python @@ -120,7 +124,15 @@ def __init__( "Use agent card to invoke agent. The agent endpoint will use the `url` in agent card." ) - self.httpx_client = httpx.AsyncClient() + scheme = httpx.URL(self.vefaas_endpoint).scheme + proxy = ( + os.getenv(f"{scheme.upper()}_PROXY") or os.getenv(f"{scheme}_proxy") or None + ) + # An explicit transport prevents HTTPX from loading unrelated proxies + # while retaining environment-based certificate configuration + self.httpx_client = httpx.AsyncClient( + transport=httpx.AsyncHTTPTransport(proxy=proxy) + ) def _get_vefaas_endpoint( self,