From 1313cf1d8e1a1a078497606b2eb6d61b9a0161bc Mon Sep 17 00:00:00 2001 From: Pramod Surya Date: Mon, 2 Nov 2026 23:58:40 +0530 Subject: [PATCH] feat(bench): Let the operator trust named private hosts for backup downloads The backup URL check refuses any host that resolves to a private address. On a self-hosted Press, the platform's own S3-compatible backup store (Garage, MinIO, Ceph RGW) often sits on the private network, so "New Site from Backup" from it fails with "The URL resolves to an address that is not public." AGENT_TRUSTED_PRIVATE_HOSTS takes exact hostnames, comma-separated. Those hosts may resolve to private addresses; every other host, including each redirect target, is checked as before. Unset, nothing changes. Co-Authored-By: Claude Opus 5.5 --- agent/remote_url.py | 15 +++++++++++++- agent/tests/test_remote_url.py | 37 ++++++++++++++++++++++++++++++++++ 2 files changed, 51 insertions(+), 1 deletion(-) diff --git a/agent/remote_url.py b/agent/remote_url.py index 9661def9..091610cd 100644 --- a/agent/remote_url.py +++ b/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: diff --git a/agent/tests/test_remote_url.py b/agent/tests/test_remote_url.py index 7afa2e19..63d5bda8 100644 --- a/agent/tests/test_remote_url.py +++ b/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