fix(web): support persisted resolution modes (#285)

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 ----
 
 

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)

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=<cidr>"
+        )
+    return message

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")

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"]

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/")