feat(bench): Let the operator trust named private hosts for backup downloads

agent/remote_url.py

@@ -3,11 +3,17 @@
 Each request goes only to a public address. The host is resolved once, and the
 connection is pinned to the address that was checked. Redirects are followed by
 hand, and each target is checked the same way.
+
+An operator can trust named hosts on a private network, such as the platform's
+own S3-compatible backup store, by listing their exact hostnames, comma-separated,
+in the AGENT_TRUSTED_PRIVATE_HOSTS environment variable. Those hosts may resolve
+to private addresses; every other host is still checked.
 """
 
 from __future__ import annotations
 
 import ipaddress
+import os
 import socket
 from contextlib import contextmanager
 from typing import TYPE_CHECKING
@@ -61,13 +67,20 @@ def _resolve_public_ip(hostname: str) -> str:
         raise RemoteURLError(f"Could not resolve the host '{hostname}'.") from error
 
     ips = [str(address[4][0]) for address in addresses]
+    trusted = hostname.lower() in _trusted_private_hosts()
     for ip in ips:
-        if not _is_public(ip):
+        if not trusted and not _is_public(ip):
             raise RemoteURLError("The URL resolves to an address that is not public.")
     # Prefer IPv4, as the server may not have IPv6 egress
     return next((ip for ip in ips if ipaddress.ip_address(ip).version == 4), ips[0])
 
 
+def _trusted_private_hosts() -> frozenset[str]:
+    """Exact hostnames the operator allows to resolve to private addresses."""
+    hosts = os.environ.get("AGENT_TRUSTED_PRIVATE_HOSTS", "")
+    return frozenset(host.strip().lower() for host in hosts.split(",") if host.strip())
+
+
 def _is_public(ip: str) -> bool:
     address = ipaddress.ip_address(ip)
     if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped:

agent/tests/test_remote_url.py

@@ -204,6 +204,43 @@ def test_download_refuses_host_with_any_address_that_is_not_public(self):
                 download_file("https://files.example.com/backup.sql.gz", prefix=self.directory)
             send.assert_not_called()
 
+    def test_download_allows_a_trusted_private_host(self):
+        send, hosts = fake_responses((200, None))
+        with patch.dict(
+            os.environ, {"AGENT_TRUSTED_PRIVATE_HOSTS": " S3.internal.example , other.example"}
+        ), patch("agent.remote_url.socket.getaddrinfo", return_value=resolve_to("10.0.0.5")), patch(
+            "requests.adapters.HTTPAdapter.send", send
+        ):
+            path = download_file("https://s3.internal.example/bucket/backup.sql.gz", prefix=self.directory)
+
+        self.assertEqual(hosts, ["s3.internal.example"])
+        with open(path, "rb") as f:
+            self.assertEqual(f.read(), b"backup")
+
+    def test_download_still_refuses_other_private_hosts_when_some_are_trusted(self):
+        with patch.dict(os.environ, {"AGENT_TRUSTED_PRIVATE_HOSTS": "s3.internal.example"}):
+            for host in ("files.internal.example", "internal.example", "s3.internal.example.attacker.com"):
+                with self.subTest(host=host), patch(
+                    "agent.remote_url.socket.getaddrinfo", return_value=resolve_to("10.0.0.5")
+                ), patch("requests.adapters.HTTPAdapter.send") as send, self.assertRaisesRegex(
+                    RemoteURLError, "not public"
+                ):
+                    download_file(f"https://{host}/backup.sql.gz", prefix=self.directory)
+                send.assert_not_called()
+
+    def test_download_checks_redirects_from_a_trusted_private_host(self):
+        send, hosts = fake_responses((302, "http://files.internal.example/backup.sql.gz"), (200, None))
+        with patch.dict(os.environ, {"AGENT_TRUSTED_PRIVATE_HOSTS": "s3.internal.example"}), patch(
+            "agent.remote_url.socket.getaddrinfo",
+            resolve_by_host({"s3.internal.example": "10.0.0.5", "files.internal.example": "10.0.0.6"}),
+        ), patch("requests.adapters.HTTPAdapter.send", send), self.assertRaisesRegex(
+            RemoteURLError, "not public"
+        ):
+            download_file("https://s3.internal.example/bucket/backup.sql.gz", prefix=self.directory)
+
+        self.assertEqual(hosts, ["s3.internal.example"])
+        self.assertEqual(os.listdir(self.directory), [])
+
     def test_download_refuses_host_that_does_not_resolve(self):
         with patch(
             "agent.remote_url.socket.getaddrinfo", side_effect=socket.gaierror