From 257916db67b1ff936ed552bffc2a4a79fa004ac7 Mon Sep 17 00:00:00 2001 From: Yang XiuYuan <31394886+patsyang@users.noreply.github.com> Date: Thu, 4 Jun 2026 09:30:31 +0800 Subject: [PATCH] fix(web): support persisted resolution modes (#285) --- src/openharness/cli.py | 68 +++++++ src/openharness/config/settings.py | 26 +++ src/openharness/utils/network_guard.py | 232 ++++++++++++++++++++-- tests/test_config/test_settings.py | 13 ++ tests/test_entrypoints/test_config_cli.py | 30 +++ tests/test_utils/test_network_guard.py | 154 ++++++++++++++ 6 files changed, 510 insertions(+), 13 deletions(-) create mode 100644 tests/test_entrypoints/test_config_cli.py create mode 100644 tests/test_utils/test_network_guard.py diff --git a/src/openharness/cli.py b/src/openharness/cli.py index 3b50d1d..1c07ba0 100644 --- a/src/openharness/cli.py +++ b/src/openharness/cli.py @@ -768,6 +768,7 @@ mcp_app = typer.Typer(name="mcp", help="Manage MCP servers") plugin_app = typer.Typer(name="plugin", help="Manage plugins") auth_app = typer.Typer(name="auth", help="Manage authentication") provider_app = typer.Typer(name="provider", help="Manage provider profiles") +config_app = typer.Typer(name="config", help="Show or update settings") cron_app = typer.Typer(name="cron", help="Manage cron scheduler and jobs") autopilot_app = typer.Typer(name="autopilot", help="Manage repo autopilot") @@ -775,6 +776,7 @@ app.add_typer(mcp_app) app.add_typer(plugin_app) app.add_typer(auth_app) app.add_typer(provider_app) +app.add_typer(config_app) app.add_typer(cron_app) app.add_typer(autopilot_app) @@ -1967,6 +1969,72 @@ def auth_copilot_logout() -> None: print("Copilot authentication cleared.") +# ---- config subcommands ---- + + +def _config_resolve_target(settings: object, key: str) -> tuple[object, str]: + target = settings + parts = key.split(".") + for part in parts[:-1]: + if not hasattr(target, part): + raise KeyError(key) + target = getattr(target, part) + leaf = parts[-1] + if not hasattr(target, leaf): + raise KeyError(key) + return target, leaf + + +def _config_coerce_value(current: object, raw: str) -> object: + if isinstance(current, bool): + lowered = raw.strip().lower() + if lowered in {"1", "true", "yes", "on"}: + return True + if lowered in {"0", "false", "no", "off"}: + return False + raise ValueError(f"Invalid boolean value: {raw}") + if isinstance(current, int) and not isinstance(current, bool): + return int(raw) + if isinstance(current, float): + return float(raw) + if isinstance(current, list): + return [entry.strip() for entry in raw.split(",") if entry.strip()] + return raw + + +@config_app.command("show") +def config_show() -> None: + """Print the resolved settings JSON.""" + from openharness.commands.registry import _settings_json_for_display + from openharness.config.settings import load_settings + + print(_settings_json_for_display(load_settings()), flush=True) + + +@config_app.command("set") +def config_set( + key: str = typer.Argument(..., help="Setting key, including dotted nested keys"), + value: str = typer.Argument(..., help="Value to store"), +) -> None: + """Persist one setting in ~/.openharness/settings.json.""" + from openharness.config.settings import load_settings, save_settings + + settings = load_settings() + try: + target, leaf = _config_resolve_target(settings, key) + except KeyError: + print(f"Unknown config key: {key}", file=sys.stderr) + raise typer.Exit(1) + try: + coerced = _config_coerce_value(getattr(target, leaf), value) + except ValueError as exc: + print(f"Error: {exc}", file=sys.stderr) + raise typer.Exit(1) + setattr(target, leaf, coerced) + save_settings(settings) + print(f"Updated {key}", flush=True) + + # ---- provider subcommands ---- diff --git a/src/openharness/config/settings.py b/src/openharness/config/settings.py index 32c5435..2c9d419 100644 --- a/src/openharness/config/settings.py +++ b/src/openharness/config/settings.py @@ -113,6 +113,14 @@ class SandboxSettings(BaseModel): docker: DockerSandboxSettings = Field(default_factory=DockerSandboxSettings) +class WebSettings(BaseModel): + """Outbound web tool configuration.""" + + proxy: str | None = None + resolution_mode: str = "auto" + synthetic_dns_cidrs: list[str] = Field(default_factory=list) + + class ProviderProfile(BaseModel): """Named provider workflow configuration.""" @@ -578,6 +586,7 @@ class Settings(BaseModel): hooks: dict[str, list[HookDefinition]] = Field(default_factory=dict) memory: MemorySettings = Field(default_factory=MemorySettings) sandbox: SandboxSettings = Field(default_factory=SandboxSettings) + web: WebSettings = Field(default_factory=WebSettings) enabled_plugins: dict[str, bool] = Field(default_factory=dict) allow_project_plugins: bool = False allow_project_skills: bool = True @@ -987,6 +996,23 @@ def _apply_env_overrides(settings: Settings) -> Settings: if sandbox_updates: updates["sandbox"] = settings.sandbox.model_copy(update=sandbox_updates) + web_updates: dict[str, Any] = {} + web_proxy = os.environ.get("OPENHARNESS_WEB_PROXY") + if web_proxy: + web_updates["proxy"] = web_proxy + web_resolution_mode = os.environ.get("OPENHARNESS_WEB_RESOLUTION_MODE") + if web_resolution_mode: + web_updates["resolution_mode"] = web_resolution_mode + web_synthetic_dns_cidrs = os.environ.get("OPENHARNESS_WEB_SYNTHETIC_DNS_CIDRS") + if web_synthetic_dns_cidrs: + web_updates["synthetic_dns_cidrs"] = [ + entry.strip() + for entry in web_synthetic_dns_cidrs.split(",") + if entry.strip() + ] + if web_updates: + updates["web"] = settings.web.model_copy(update=web_updates) + if not updates: return settings return settings.model_copy(update=updates) diff --git a/src/openharness/utils/network_guard.py b/src/openharness/utils/network_guard.py index cc82037..4fd1dde 100644 --- a/src/openharness/utils/network_guard.py +++ b/src/openharness/utils/network_guard.py @@ -4,9 +4,9 @@ from __future__ import annotations import asyncio import ipaddress -import os import socket -from urllib.parse import urljoin, urlparse +from enum import Enum +from urllib.parse import ParseResult, urljoin, urlparse import httpx @@ -16,6 +16,31 @@ _DEFAULT_PORTS = { "https": 443, } _IPAddress = ipaddress.IPv4Address | ipaddress.IPv6Address +_IPNetwork = ipaddress.IPv4Network | ipaddress.IPv6Network +_SYNTHETIC_DNS_CIDRS_SETTING = "web.synthetic_dns_cidrs" +_RESOLUTION_MODE_SETTING = "web.resolution_mode" +_PROXY_SETTING = "web.proxy" +_LOCAL_HOSTNAMES = { + "localhost", + "localhost.localdomain", + "metadata.google.internal", +} +_LOCAL_HOST_SUFFIXES = ( + ".localhost", + ".local", + ".localdomain", + ".internal", + ".cluster.local", +) + + +class ResolutionMode(str, Enum): + """How outbound web tools should interpret target DNS resolution.""" + + AUTO = "auto" + DIRECT = "direct" + PROXY = "proxy" + SYNTHETIC_DNS = "synthetic_dns" class NetworkGuardError(ValueError): @@ -33,22 +58,73 @@ def validate_http_url(url: str) -> None: raise NetworkGuardError("URLs with embedded credentials are not allowed") +def get_web_resolution_mode( + proxy: str | None = None, + *, + configured_mode: str | None = None, +) -> ResolutionMode: + """Resolve the configured web target validation mode.""" + raw_mode = (configured_mode or "").strip().lower().replace("-", "_") + if not raw_mode or raw_mode == ResolutionMode.AUTO.value: + return ResolutionMode.PROXY if proxy else ResolutionMode.DIRECT + try: + mode = ResolutionMode(raw_mode) + except ValueError as exc: + allowed = ", ".join(mode.value for mode in ResolutionMode) + raise NetworkGuardError(f"{_RESOLUTION_MODE_SETTING} must be one of: {allowed}") from exc + if mode is ResolutionMode.AUTO: + return ResolutionMode.PROXY if proxy else ResolutionMode.DIRECT + if mode is ResolutionMode.PROXY and not proxy: + raise NetworkGuardError(f"{_RESOLUTION_MODE_SETTING}=proxy requires {_PROXY_SETTING}") + return mode + + +def parse_synthetic_dns_cidrs(value: str | None = None) -> tuple[_IPNetwork, ...]: + """Parse user-declared synthetic DNS CIDRs.""" + raw_value = "" if value is None else value + entries = [entry.strip() for entry in raw_value.split(",") if entry.strip()] + networks: list[_IPNetwork] = [] + for entry in entries: + try: + networks.append(ipaddress.ip_network(entry, strict=False)) + except ValueError as exc: + raise NetworkGuardError(f"invalid {_SYNTHETIC_DNS_CIDRS_SETTING} entry: {entry}") from exc + return tuple(networks) + + async def ensure_public_http_url(url: str) -> None: """Reject loopback, private-network, and other non-public HTTP targets.""" - validate_http_url(url) - parsed = urlparse(url) - assert parsed.hostname is not None # covered by validate_http_url + parsed = _validated_parsed_http_url(url) + hostname = _normalized_hostname(parsed.hostname) + literal = _parse_ip_literal(hostname) + if literal is not None: + _ensure_global_literal_ip(literal) + return + _ensure_not_local_hostname(hostname) port = parsed.port or _DEFAULT_PORTS[parsed.scheme] - addresses = await _resolve_host_addresses(parsed.hostname, port) + addresses = await _resolve_host_addresses(hostname, port) if not addresses: - raise NetworkGuardError(f"target host did not resolve: {parsed.hostname}") + raise NetworkGuardError(f"target host did not resolve: {hostname}") blocked = sorted({str(address) for address in addresses if not address.is_global}) if blocked: - rendered = ", ".join(blocked[:3]) - if len(blocked) > 3: - rendered += ", ..." - raise NetworkGuardError(f"target resolves to non-public address(es): {rendered}") + raise NetworkGuardError(_format_blocked_addresses(blocked, include_synthetic_dns_hint=True)) + + +async def ensure_http_url_allowed( + url: str, + *, + mode: ResolutionMode, + synthetic_cidrs: tuple[_IPNetwork, ...] = (), +) -> None: + """Validate one outbound URL according to the configured resolution mode.""" + if mode is ResolutionMode.DIRECT: + await ensure_public_http_url(url) + return + if mode is ResolutionMode.PROXY: + _ensure_proxy_safe_http_url(url) + return + await _ensure_synthetic_dns_safe_http_url(url, synthetic_cidrs=synthetic_cidrs) async def fetch_public_http_response( @@ -64,9 +140,19 @@ async def fetch_public_http_response( current_url = url current_params = params - resolved_proxy = proxy if proxy is not None else os.environ.get("OPENHARNESS_WEB_PROXY") + web_settings = _load_configured_web_settings() + resolved_proxy = proxy if proxy is not None else web_settings.proxy if resolved_proxy: validate_http_url(resolved_proxy) + mode = get_web_resolution_mode( + resolved_proxy, + configured_mode=web_settings.resolution_mode, + ) + synthetic_cidrs = ( + parse_synthetic_dns_cidrs(",".join(web_settings.synthetic_dns_cidrs)) + if mode is ResolutionMode.SYNTHETIC_DNS + else () + ) async with httpx.AsyncClient( follow_redirects=False, @@ -75,7 +161,11 @@ async def fetch_public_http_response( proxy=resolved_proxy, ) as client: for redirect_count in range(max_redirects + 1): - await ensure_public_http_url(current_url) + await ensure_http_url_allowed( + current_url, + mode=mode, + synthetic_cidrs=synthetic_cidrs, + ) response = await client.get( current_url, params=current_params, @@ -96,6 +186,75 @@ async def fetch_public_http_response( raise NetworkGuardError("request failed before receiving a response") +class _ConfiguredWebSettings: + def __init__( + self, + *, + proxy: str | None, + resolution_mode: str, + synthetic_dns_cidrs: list[str], + ) -> None: + self.proxy = proxy + self.resolution_mode = resolution_mode + self.synthetic_dns_cidrs = synthetic_dns_cidrs + + +def _load_configured_web_settings() -> _ConfiguredWebSettings: + """Load persisted web settings, including environment overrides.""" + from openharness.config import load_settings + + web = load_settings().web + return _ConfiguredWebSettings( + proxy=web.proxy, + resolution_mode=web.resolution_mode, + synthetic_dns_cidrs=list(web.synthetic_dns_cidrs), + ) + + +def _ensure_proxy_safe_http_url(url: str) -> None: + """Validate a URL whose hostname will be resolved by an explicit proxy.""" + parsed = _validated_parsed_http_url(url) + hostname = _normalized_hostname(parsed.hostname) + literal = _parse_ip_literal(hostname) + if literal is not None: + _ensure_global_literal_ip(literal) + return + _ensure_not_local_hostname(hostname) + + +async def _ensure_synthetic_dns_safe_http_url( + url: str, + *, + synthetic_cidrs: tuple[_IPNetwork, ...], +) -> None: + """Validate a URL in a user-declared synthetic DNS environment.""" + if not synthetic_cidrs: + raise NetworkGuardError( + f"{ResolutionMode.SYNTHETIC_DNS.value} mode requires {_SYNTHETIC_DNS_CIDRS_SETTING}" + ) + parsed = _validated_parsed_http_url(url) + hostname = _normalized_hostname(parsed.hostname) + literal = _parse_ip_literal(hostname) + if literal is not None: + _ensure_global_literal_ip(literal) + return + _ensure_not_local_hostname(hostname) + port = parsed.port or _DEFAULT_PORTS[parsed.scheme] + addresses = await _resolve_host_addresses(hostname, port) + if not addresses: + raise NetworkGuardError(f"target host did not resolve: {hostname}") + + blocked = sorted( + { + str(address) + for address in addresses + if not address.is_global and not _address_in_networks(address, synthetic_cidrs) + } + ) + if blocked: + raise NetworkGuardError(_format_blocked_addresses(blocked)) + + async def _resolve_host_addresses(host: str, port: int) -> set[_IPAddress]: """Resolve a host into concrete IP addresses.""" literal = _parse_ip_literal(host) @@ -121,6 +280,8 @@ async def _resolve_host_addresses(host: str, port: int) -> set[_IPAddress]: candidate = sockaddr[0] else: continue + if not isinstance(candidate, str): + continue parsed = _parse_ip_literal(candidate) if parsed is not None: addresses.add(parsed) @@ -132,3 +293,48 @@ def _parse_ip_literal(value: str) -> _IPAddress | None: return ipaddress.ip_address(value) except ValueError: return None + + +def _validated_parsed_http_url(url: str) -> ParseResult: + validate_http_url(url) + parsed = urlparse(url) + assert parsed.hostname is not None # covered by validate_http_url + return parsed + + +def _normalized_hostname(hostname: str | None) -> str: + assert hostname is not None # covered by validate_http_url + return hostname.rstrip(".").lower() + + +def _ensure_global_literal_ip(address: _IPAddress) -> None: + if not address.is_global: + raise NetworkGuardError(f"target resolves to non-public address(es): {address}") + + +def _ensure_not_local_hostname(hostname: str) -> None: + if hostname in _LOCAL_HOSTNAMES or any(hostname.endswith(suffix) for suffix in _LOCAL_HOST_SUFFIXES): + raise NetworkGuardError(f"local hostnames are not allowed: {hostname}") + if "." not in hostname: + raise NetworkGuardError(f"single-label hostnames are not allowed: {hostname}") + + +def _address_in_networks(address: _IPAddress, networks: tuple[_IPNetwork, ...]) -> bool: + return any(address.version == network.version and address in network for network in networks) + + +def _format_blocked_addresses( + blocked: list[str], + *, + include_synthetic_dns_hint: bool = False, +) -> str: + rendered = ", ".join(blocked[:3]) + if len(blocked) > 3: + rendered += ", ..." + message = f"target resolves to non-public address(es): {rendered}" + if include_synthetic_dns_hint: + message += ( + "; if this domain intentionally resolves through synthetic DNS, configure " + "web.resolution_mode=synthetic_dns and web.synthetic_dns_cidrs=" + ) + return message diff --git a/tests/test_config/test_settings.py b/tests/test_config/test_settings.py index c79c09d..33b07ca 100644 --- a/tests/test_config/test_settings.py +++ b/tests/test_config/test_settings.py @@ -32,6 +32,8 @@ class TestSettings: assert s.permission.mode == "default" assert s.sandbox.enabled is False assert s.sandbox.filesystem.allow_write == ["."] + assert s.web.resolution_mode == "auto" + assert s.web.synthetic_dns_cidrs == [] def test_resolve_api_key_from_instance(self): s = Settings(api_key="sk-test-123") @@ -78,6 +80,17 @@ class TestSettings: assert updated.permission.mode == "full_auto" assert s.permission.mode == "default" + def test_web_settings_env_overrides(self, monkeypatch): + monkeypatch.setenv("OPENHARNESS_WEB_PROXY", "http://proxy.example.com:7890") + monkeypatch.setenv("OPENHARNESS_WEB_RESOLUTION_MODE", "synthetic_dns") + monkeypatch.setenv("OPENHARNESS_WEB_SYNTHETIC_DNS_CIDRS", "100.64.0.0/10,203.0.113.0/24") + + updated = _apply_env_overrides(Settings()) + + assert updated.web.proxy == "http://proxy.example.com:7890" + assert updated.web.resolution_mode == "synthetic_dns" + assert updated.web.synthetic_dns_cidrs == ["100.64.0.0/10", "203.0.113.0/24"] + def test_merge_cli_overrides_returns_new_instance(self): s = Settings() updated = s.merge_cli_overrides(model="claude-opus-4-20250514") diff --git a/tests/test_entrypoints/test_config_cli.py b/tests/test_entrypoints/test_config_cli.py new file mode 100644 index 0000000..4e3f293 --- /dev/null +++ b/tests/test_entrypoints/test_config_cli.py @@ -0,0 +1,30 @@ +"""Tests for the top-level config CLI.""" + +from __future__ import annotations + +from pathlib import Path + +from typer.testing import CliRunner + +from openharness.cli import app +from openharness.config.settings import load_settings + + +def test_cli_config_set_persists_nested_web_settings(tmp_path: Path, monkeypatch): + monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) + runner = CliRunner() + + mode_result = runner.invoke(app, ["config", "set", "web.resolution_mode", "synthetic_dns"]) + assert mode_result.exit_code == 0 + assert "Updated web.resolution_mode" in mode_result.output + + cidrs_result = runner.invoke( + app, + ["config", "set", "web.synthetic_dns_cidrs", "100.64.0.0/10,203.0.113.0/24"], + ) + assert cidrs_result.exit_code == 0 + assert "Updated web.synthetic_dns_cidrs" in cidrs_result.output + + settings = load_settings() + assert settings.web.resolution_mode == "synthetic_dns" + assert settings.web.synthetic_dns_cidrs == ["100.64.0.0/10", "203.0.113.0/24"] diff --git a/tests/test_utils/test_network_guard.py b/tests/test_utils/test_network_guard.py new file mode 100644 index 0000000..afa672b --- /dev/null +++ b/tests/test_utils/test_network_guard.py @@ -0,0 +1,154 @@ +"""Tests for outbound HTTP target validation.""" + +from __future__ import annotations + +import ipaddress + +import httpx +import pytest + +from openharness.utils.network_guard import fetch_public_http_response + + +class FakeAsyncClient: + def __init__(self, **kwargs: object) -> None: + self.kwargs = kwargs + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return None + + async def get(self, url: str, **kwargs: object) -> httpx.Response: + request = httpx.Request("GET", url, params=kwargs.get("params")) + return httpx.Response(200, text="ok", request=request) + + +@pytest.fixture(autouse=True) +def isolated_config_dir(tmp_path, monkeypatch): + monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) + + +@pytest.mark.asyncio +async def test_fetch_public_http_response_direct_rejects_non_public_dns(monkeypatch): + async def fake_resolve(host: str, port: int): + return {ipaddress.ip_address("100.64.1.2")} + + monkeypatch.setattr("openharness.utils.network_guard._resolve_host_addresses", fake_resolve) + + with pytest.raises(ValueError, match="synthetic DNS"): + await fetch_public_http_response("https://example.com/") + + +@pytest.mark.asyncio +async def test_fetch_public_http_response_synthetic_dns_allows_declared_cidr(monkeypatch): + async def fake_resolve(host: str, port: int): + return {ipaddress.ip_address("100.64.1.2")} + + monkeypatch.setenv("OPENHARNESS_WEB_RESOLUTION_MODE", "synthetic_dns") + monkeypatch.setenv("OPENHARNESS_WEB_SYNTHETIC_DNS_CIDRS", "100.64.0.0/10") + monkeypatch.setattr("openharness.utils.network_guard._resolve_host_addresses", fake_resolve) + monkeypatch.setattr(httpx, "AsyncClient", FakeAsyncClient) + + response = await fetch_public_http_response("https://example.com/") + + assert response.status_code == 200 + + +@pytest.mark.asyncio +async def test_fetch_public_http_response_synthetic_dns_uses_persisted_settings( + monkeypatch, +): + from openharness.config.settings import Settings, WebSettings, save_settings + + async def fake_resolve(host: str, port: int): + return {ipaddress.ip_address("100.64.1.2")} + + save_settings( + Settings( + web=WebSettings( + resolution_mode="synthetic_dns", + synthetic_dns_cidrs=["100.64.0.0/10"], + ) + ) + ) + monkeypatch.setattr("openharness.utils.network_guard._resolve_host_addresses", fake_resolve) + monkeypatch.setattr(httpx, "AsyncClient", FakeAsyncClient) + + response = await fetch_public_http_response("https://example.com/") + + assert response.status_code == 200 + + +@pytest.mark.asyncio +async def test_fetch_public_http_response_synthetic_dns_requires_declared_cidrs(monkeypatch): + monkeypatch.setenv("OPENHARNESS_WEB_RESOLUTION_MODE", "synthetic_dns") + + with pytest.raises(ValueError, match="web.synthetic_dns_cidrs"): + await fetch_public_http_response("https://example.com/") + + +@pytest.mark.asyncio +async def test_fetch_public_http_response_synthetic_dns_rejects_literal_non_public_ip( + monkeypatch, +): + monkeypatch.setenv("OPENHARNESS_WEB_RESOLUTION_MODE", "synthetic_dns") + monkeypatch.setenv("OPENHARNESS_WEB_SYNTHETIC_DNS_CIDRS", "100.64.0.0/10") + + with pytest.raises(ValueError, match="non-public"): + await fetch_public_http_response("http://100.64.1.2/") + + +@pytest.mark.asyncio +async def test_fetch_public_http_response_synthetic_dns_rejects_undeclared_private_dns( + monkeypatch, +): + async def fake_resolve(host: str, port: int): + return {ipaddress.ip_address("10.0.0.1")} + + monkeypatch.setenv("OPENHARNESS_WEB_RESOLUTION_MODE", "synthetic_dns") + monkeypatch.setenv("OPENHARNESS_WEB_SYNTHETIC_DNS_CIDRS", "100.64.0.0/10") + monkeypatch.setattr("openharness.utils.network_guard._resolve_host_addresses", fake_resolve) + + with pytest.raises(ValueError, match="non-public"): + await fetch_public_http_response("https://example.com/") + + +@pytest.mark.asyncio +async def test_fetch_public_http_response_proxy_mode_does_not_resolve_target_dns(monkeypatch): + async def fail_resolve(host: str, port: int): + raise AssertionError("proxy mode should not resolve ordinary target domains locally") + + monkeypatch.setenv("OPENHARNESS_WEB_PROXY", "http://proxy.example.com:7890") + monkeypatch.setenv("OPENHARNESS_WEB_SYNTHETIC_DNS_CIDRS", "not-a-cidr") + monkeypatch.setattr("openharness.utils.network_guard._resolve_host_addresses", fail_resolve) + monkeypatch.setattr(httpx, "AsyncClient", FakeAsyncClient) + + response = await fetch_public_http_response("https://example.com/") + + assert response.status_code == 200 + + +@pytest.mark.asyncio +async def test_fetch_public_http_response_proxy_mode_rejects_literal_non_public_ip(monkeypatch): + monkeypatch.setenv("OPENHARNESS_WEB_PROXY", "http://proxy.example.com:7890") + + with pytest.raises(ValueError, match="non-public"): + await fetch_public_http_response("http://127.0.0.1/") + + +@pytest.mark.asyncio +async def test_fetch_public_http_response_rejects_non_public_redirect_in_proxy_mode( + monkeypatch, +): + class RedirectClient(FakeAsyncClient): + async def get(self, url: str, **kwargs: object) -> httpx.Response: + request = httpx.Request("GET", url, params=kwargs.get("params")) + return httpx.Response(302, headers={"Location": "http://127.0.0.1/"}, request=request) + + monkeypatch.setenv("OPENHARNESS_WEB_PROXY", "http://proxy.example.com:7890") + monkeypatch.setattr(httpx, "AsyncClient", RedirectClient) + + with pytest.raises(ValueError, match="non-public"): + await fetch_public_http_response("https://example.com/")