diff --git a/src/govoplan_core/security/http_fetch.py b/src/govoplan_core/security/http_fetch.py index 82625b0..974b939 100644 --- a/src/govoplan_core/security/http_fetch.py +++ b/src/govoplan_core/security/http_fetch.py @@ -5,6 +5,12 @@ import urllib.request from dataclasses import dataclass from typing import Mapping +from govoplan_core.security.outbound_http import ( + bounded_response_bytes, + build_outbound_http_opener, + validate_outbound_http_url, +) + @dataclass(frozen=True, slots=True) class HttpFetchResponse: @@ -40,17 +46,26 @@ def fetch_http( label: str = "URL", method: str = "GET", headers: Mapping[str, str] | None = None, + max_bytes: int | None = None, ) -> HttpFetchResponse: + validated_url = validate_outbound_http_url(url, label=label) request = urllib.request.Request( # noqa: S310 - URL is restricted to validated HTTP(S). - validate_http_url(url, label=label), + validated_url, headers=dict(headers or {}), method=method, ) - with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310 - URL is validated by validate_http_url. # nosec B310 # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected + opener = build_outbound_http_opener(_PolicyRedirectHandler(label=label)) + with opener.open(request, timeout=timeout) as response: # noqa: S310 - URL and every redirect are policy-validated. # nosec B310 # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected + response_headers = dict(response.headers.items()) return HttpFetchResponse( status=int(getattr(response, "status", 0)), - headers=dict(response.headers.items()), - body=response.read(), + headers=response_headers, + body=bounded_response_bytes( + response, + headers=response_headers, + max_bytes=max_bytes, + label=f"{label} response", + ), ) @@ -62,5 +77,29 @@ def fetch_http_text( method: str = "GET", headers: Mapping[str, str] | None = None, encoding: str = "utf-8", + max_bytes: int | None = None, ) -> str: - return fetch_http(url, timeout=timeout, label=label, method=method, headers=headers).text(encoding) + return fetch_http(url, timeout=timeout, label=label, method=method, headers=headers, max_bytes=max_bytes).text(encoding) + + +class _PolicyRedirectHandler(urllib.request.HTTPRedirectHandler): + def __init__(self, *, label: str) -> None: + super().__init__() + self._label = label + + def redirect_request(self, req, fp, code, msg, headers, newurl): # type: ignore[no-untyped-def] + candidate = validate_outbound_http_url(newurl, label=f"{self._label} redirect") + previous = urllib.parse.urlparse(req.full_url) + redirected = urllib.parse.urlparse(candidate) + if previous.scheme.lower() == "https" and redirected.scheme.lower() != "https": + return None + new_request = super().redirect_request(req, fp, code, msg, headers, candidate) + if new_request is not None and _http_origin(previous) != _http_origin(redirected): + for header in ("Authorization", "Proxy-Authorization", "Cookie", "Cookie2"): + new_request.remove_header(header) + return new_request + + +def _http_origin(parsed: urllib.parse.ParseResult) -> tuple[str, str, int]: + scheme = parsed.scheme.lower() + return scheme, (parsed.hostname or "").lower(), parsed.port or (443 if scheme == "https" else 80) diff --git a/src/govoplan_core/security/outbound_http.py b/src/govoplan_core/security/outbound_http.py new file mode 100644 index 0000000..0c57fec --- /dev/null +++ b/src/govoplan_core/security/outbound_http.py @@ -0,0 +1,366 @@ +from __future__ import annotations + +import ipaddress +import errno +import http.client +import os +import socket +import urllib.parse +import urllib.request +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +from typing import BinaryIO, Final + + +DEFAULT_STRUCTURED_RESPONSE_BYTES: Final = 16 * 1024 * 1024 +DEFAULT_FILE_TRANSFER_BYTES: Final = 512 * 1024 * 1024 +_READ_CHUNK_BYTES: Final = 64 * 1024 +_TRUE_VALUES: Final = frozenset({"1", "true", "yes", "on"}) +_FALSE_VALUES: Final = frozenset({"0", "false", "no", "off"}) + + +class OutboundHttpError(RuntimeError): + """Base error for deployment-wide outbound HTTP policy failures.""" + + +class OutboundHttpBlocked(OutboundHttpError): + """Raised when a URL targets an address forbidden by deployment policy.""" + + +class OutboundResponseTooLarge(OutboundHttpError): + """Raised before a connector can retain an oversized remote response.""" + + +@dataclass(frozen=True, slots=True) +class OutboundHttpPolicy: + allow_private_networks: bool + structured_response_bytes: int + file_transfer_bytes: int + + +def outbound_http_policy(environ: Mapping[str, str] | None = None) -> OutboundHttpPolicy: + env = os.environ if environ is None else environ + return OutboundHttpPolicy( + allow_private_networks=_private_network_default(env), + structured_response_bytes=_positive_int( + env.get("GOVOPLAN_CONNECTOR_MAX_STRUCTURED_RESPONSE_BYTES"), + default=DEFAULT_STRUCTURED_RESPONSE_BYTES, + name="GOVOPLAN_CONNECTOR_MAX_STRUCTURED_RESPONSE_BYTES", + ), + file_transfer_bytes=_positive_int( + env.get("GOVOPLAN_CONNECTOR_MAX_FILE_TRANSFER_BYTES"), + default=DEFAULT_FILE_TRANSFER_BYTES, + name="GOVOPLAN_CONNECTOR_MAX_FILE_TRANSFER_BYTES", + ), + ) + + +def validate_outbound_http_url( + value: str, + *, + label: str = "Connector URL", + policy: OutboundHttpPolicy | None = None, +) -> str: + parsed = urllib.parse.urlparse(str(value).strip()) + if parsed.scheme.lower() not in {"http", "https"} or not parsed.netloc or not parsed.hostname: + raise OutboundHttpBlocked(f"{label} must be an absolute HTTP(S) URL") + if parsed.username or parsed.password: + raise OutboundHttpBlocked(f"{label} must not include embedded credentials") + try: + port = parsed.port or (443 if parsed.scheme.lower() == "https" else 80) + except ValueError as exc: + raise OutboundHttpBlocked(f"{label} has an invalid port") from exc + validate_outbound_host(parsed.hostname, port=port, label=label, policy=policy) + return urllib.parse.urlunparse(parsed) + + +def validate_unpinned_sdk_http_url( + value: str, + *, + label: str, + policy: OutboundHttpPolicy | None = None, +) -> str: + """Fail closed when an SDK cannot connect to a prevalidated DNS answer.""" + + active_policy = policy or outbound_http_policy() + validated = validate_outbound_http_url(value, label=label, policy=active_policy) + if active_policy.allow_private_networks: + return validated + raise OutboundHttpBlocked( + f"{label} uses an SDK that cannot pin connection peers or revalidate every SDK-managed redirect; " + "it is disabled while application-enforced public-only egress is required" + ) + + +def validate_outbound_host( + hostname: str, + *, + port: int, + label: str = "Connector host", + policy: OutboundHttpPolicy | None = None, +) -> tuple[str, ...]: + active_policy = policy or outbound_http_policy() + host = str(hostname).strip().rstrip(".") + if not host: + raise OutboundHttpBlocked(f"{label} must include a hostname") + if not 1 <= int(port) <= 65535: + raise OutboundHttpBlocked(f"{label} has an invalid port") + if active_policy.allow_private_networks: + return () + try: + results = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM) + except socket.gaierror as exc: + raise OutboundHttpBlocked(f"{label} hostname could not be resolved") from exc + addresses = tuple(dict.fromkeys(str(item[4][0]).split("%", 1)[0] for item in results if item[4])) + if not addresses: + raise OutboundHttpBlocked(f"{label} hostname did not resolve to an address") + forbidden = [address for address in addresses if not _is_public_address(address)] + if forbidden: + raise OutboundHttpBlocked( + f"{label} resolves to a non-public network; " + "set GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS=true for this deployment to permit it" + ) + return addresses + + +def create_outbound_connection( + hostname: str, + port: int, + timeout: float | object | None = None, + source_address: tuple[str, int] | None = None, + socket_options: Iterable[tuple[object, ...]] | None = None, + *, + label: str = "Connector host", + policy: OutboundHttpPolicy | None = None, +) -> socket.socket: + """Resolve, validate, and connect to the exact approved address records. + + Hostname resolution happens exactly once for this connection attempt. The + returned socket connects directly to one of those validated sockaddr + records, while higher protocol layers retain the original hostname for + HTTP Host, TLS SNI, and certificate verification. + """ + + active_policy = policy or outbound_http_policy() + records = _resolved_address_records(hostname, port=port, label=label, policy=active_policy) + effective_timeout = None if timeout is socket._GLOBAL_DEFAULT_TIMEOUT else timeout # type: ignore[attr-defined] + last_error: OSError | None = None + for family, socktype, proto, _canonname, sockaddr in records: + sock: socket.socket | None = None + try: + sock = socket.socket(family, socktype, proto) + sock.settimeout(effective_timeout) # type: ignore[arg-type] + if source_address is not None: + bind_address: tuple[object, ...] = source_address + if family == socket.AF_INET6 and len(source_address) == 2: + bind_address = (source_address[0], source_address[1], 0, 0) + sock.bind(bind_address) + for option in socket_options or (): + sock.setsockopt(*option) + try: + sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + except OSError as exc: + if exc.errno != errno.ENOPROTOOPT: + raise + sock.connect(sockaddr) + return sock + except OSError as exc: + last_error = exc + if sock is not None: + sock.close() + if last_error is not None: + raise last_error + raise OutboundHttpBlocked(f"{label} hostname did not resolve to a usable address") + + +def pinned_outbound_hostname( + hostname: str, + *, + port: int, + label: str = "Connector host", + policy: OutboundHttpPolicy | None = None, +) -> str: + """Return the original allowed host or a public numeric connection target.""" + + active_policy = policy or outbound_http_policy() + host = str(hostname).strip().rstrip(".") + if active_policy.allow_private_networks: + return host + records = _resolved_address_records(host, port=port, label=label, policy=active_policy) + addresses = tuple(dict.fromkeys(str(item[4][0]).split("%", 1)[0] for item in records if item[4])) + ipv4 = next((address for address in addresses if ":" not in address), None) + return ipv4 or addresses[0] + + +def build_outbound_http_opener(*handlers: urllib.request.BaseHandler) -> urllib.request.OpenerDirector: + """Build a proxy-free urllib opener whose sockets use approved addresses.""" + + return urllib.request.build_opener( + urllib.request.ProxyHandler({}), + _OutboundHTTPHandler(), + _OutboundHTTPSHandler(), + *handlers, + ) + + +def response_limit(kind: str, *, policy: OutboundHttpPolicy | None = None) -> int: + active_policy = policy or outbound_http_policy() + if kind == "structured": + return active_policy.structured_response_bytes + if kind == "file": + return active_policy.file_transfer_bytes + raise ValueError("Response kind must be 'structured' or 'file'") + + +def bounded_response_bytes( + stream: BinaryIO, + *, + headers: Mapping[str, str] | None = None, + max_bytes: int | None = None, + kind: str = "structured", + label: str = "Connector response", +) -> bytes: + configured_limit = response_limit(kind) + limit = configured_limit if max_bytes is None else min(int(max_bytes), configured_limit) + if limit <= 0: + raise ValueError("max_bytes must be positive") + declared_size = _content_length(headers or {}) + if declared_size is not None and declared_size > limit: + raise OutboundResponseTooLarge(f"{label} exceeds the configured limit of {limit} bytes") + body = bytearray() + while len(body) <= limit: + chunk = stream.read(min(_READ_CHUNK_BYTES, limit + 1 - len(body))) + if not chunk: + return bytes(body) + body.extend(chunk) + raise OutboundResponseTooLarge(f"{label} exceeds the configured limit of {limit} bytes") + + +def bounded_chunks_bytes( + chunks: Iterable[bytes], + *, + headers: Mapping[str, str] | None = None, + max_bytes: int | None = None, + kind: str = "structured", + label: str = "Connector response", +) -> bytes: + configured_limit = response_limit(kind) + limit = configured_limit if max_bytes is None else min(int(max_bytes), configured_limit) + if limit <= 0: + raise ValueError("max_bytes must be positive") + declared_size = _content_length(headers or {}) + if declared_size is not None and declared_size > limit: + raise OutboundResponseTooLarge(f"{label} exceeds the configured limit of {limit} bytes") + body = bytearray() + for chunk in chunks: + if len(chunk) > limit - len(body): + raise OutboundResponseTooLarge(f"{label} exceeds the configured limit of {limit} bytes") + body.extend(chunk) + return bytes(body) + + +def _private_network_default(environ: Mapping[str, str]) -> bool: + configured = environ.get("GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS") + if configured is not None and configured.strip(): + value = configured.strip().lower() + if value in _TRUE_VALUES: + return True + if value in _FALSE_VALUES: + return False + raise ValueError("GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS must be true or false") + return environ.get("APP_ENV", "dev").strip().lower() in {"dev", "development", "test"} + + +def _positive_int(value: str | None, *, default: int, name: str) -> int: + if value is None or not value.strip(): + return default + try: + parsed = int(value) + except ValueError as exc: + raise ValueError(f"{name} must be a positive integer") from exc + if parsed <= 0: + raise ValueError(f"{name} must be a positive integer") + return parsed + + +def _content_length(headers: Mapping[str, str]) -> int | None: + value = next((item for key, item in headers.items() if key.casefold() == "content-length"), None) + if value is None: + return None + try: + parsed = int(value) + except (TypeError, ValueError): + return None + return parsed if parsed >= 0 else None + + +def _is_public_address(value: str) -> bool: + try: + return ipaddress.ip_address(value).is_global + except ValueError: + return False + + +def _resolved_address_records( + hostname: str, + *, + port: int, + label: str, + policy: OutboundHttpPolicy, +) -> tuple[tuple[int, int, int, str, tuple[object, ...]], ...]: + host = str(hostname).strip().rstrip(".") + if not host: + raise OutboundHttpBlocked(f"{label} must include a hostname") + if not 1 <= int(port) <= 65535: + raise OutboundHttpBlocked(f"{label} has an invalid port") + try: + results = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM) + except socket.gaierror as exc: + raise OutboundHttpBlocked(f"{label} hostname could not be resolved") from exc + records = tuple(results) + if not records: + raise OutboundHttpBlocked(f"{label} hostname did not resolve to an address") + if not policy.allow_private_networks: + addresses = tuple(str(item[4][0]).split("%", 1)[0] for item in records if item[4]) + if not addresses or any(not _is_public_address(address) for address in addresses): + raise OutboundHttpBlocked( + f"{label} resolves to a non-public network; " + "set GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS=true for this deployment to permit it" + ) + return records + + +def _create_outbound_connection_compat( + address: tuple[str, int], + timeout: float | object | None = socket._GLOBAL_DEFAULT_TIMEOUT, # type: ignore[attr-defined] + source_address: tuple[str, int] | None = None, +) -> socket.socket: + return create_outbound_connection( + address[0], + address[1], + timeout=timeout, + source_address=source_address, + label="Outbound HTTP connection", + ) + + +class _OutboundHTTPConnection(http.client.HTTPConnection): + def __init__(self, *args: object, **kwargs: object) -> None: + super().__init__(*args, **kwargs) # type: ignore[arg-type] + self._create_connection = _create_outbound_connection_compat + + +class _OutboundHTTPSConnection(http.client.HTTPSConnection): + def __init__(self, *args: object, **kwargs: object) -> None: + super().__init__(*args, **kwargs) # type: ignore[arg-type] + self._create_connection = _create_outbound_connection_compat + + +class _OutboundHTTPHandler(urllib.request.HTTPHandler): + def http_open(self, req): # type: ignore[no-untyped-def] + return self.do_open(_OutboundHTTPConnection, req) + + +class _OutboundHTTPSHandler(urllib.request.HTTPSHandler): + def https_open(self, req): # type: ignore[no-untyped-def] + return self.do_open(_OutboundHTTPSConnection, req, context=self._context) diff --git a/tests/test_http_fetch.py b/tests/test_http_fetch.py index ad5ff86..112d11c 100644 --- a/tests/test_http_fetch.py +++ b/tests/test_http_fetch.py @@ -1,8 +1,22 @@ from __future__ import annotations +import io import unittest +from unittest.mock import patch -from govoplan_core.security.http_fetch import is_http_url, validate_http_url +from govoplan_core.security.http_fetch import _PolicyRedirectHandler, is_http_url, validate_http_url +from govoplan_core.security.outbound_http import ( + DEFAULT_FILE_TRANSFER_BYTES, + DEFAULT_STRUCTURED_RESPONSE_BYTES, + OutboundHttpBlocked, + OutboundResponseTooLarge, + bounded_chunks_bytes, + bounded_response_bytes, + create_outbound_connection, + outbound_http_policy, + validate_outbound_http_url, + validate_unpinned_sdk_http_url, +) class HttpFetchTests(unittest.TestCase): @@ -22,6 +36,124 @@ class HttpFetchTests(unittest.TestCase): with self.assertRaises(ValueError): validate_http_url(value) + def test_outbound_policy_has_practical_bounded_defaults(self) -> None: + policy = outbound_http_policy({"APP_ENV": "production"}) + self.assertFalse(policy.allow_private_networks) + self.assertEqual(DEFAULT_STRUCTURED_RESPONSE_BYTES, policy.structured_response_bytes) + self.assertEqual(DEFAULT_FILE_TRANSFER_BYTES, policy.file_transfer_bytes) + + configured = outbound_http_policy( + { + "APP_ENV": "production", + "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true", + "GOVOPLAN_CONNECTOR_MAX_STRUCTURED_RESPONSE_BYTES": "1024", + "GOVOPLAN_CONNECTOR_MAX_FILE_TRANSFER_BYTES": "2048", + } + ) + self.assertTrue(configured.allow_private_networks) + self.assertEqual(1024, configured.structured_response_bytes) + self.assertEqual(2048, configured.file_transfer_bytes) + + def test_private_network_policy_is_deployment_wide(self) -> None: + denied_policy = outbound_http_policy({"APP_ENV": "production"}) + with patch( + "govoplan_core.security.outbound_http.socket.getaddrinfo", + return_value=[(2, 1, 6, "", ("127.0.0.1", 443))], + ), self.assertRaisesRegex(OutboundHttpBlocked, "non-public network"): + validate_outbound_http_url("https://connector.example.test/path", policy=denied_policy) + + allowed_policy = outbound_http_policy( + {"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true"} + ) + self.assertEqual( + "https://127.0.0.1/path", + validate_outbound_http_url("https://127.0.0.1/path", policy=allowed_policy), + ) + + def test_outbound_policy_rejects_mixed_public_and_private_dns_answers(self) -> None: + policy = outbound_http_policy({"APP_ENV": "production"}) + with patch( + "govoplan_core.security.outbound_http.socket.getaddrinfo", + return_value=[ + (2, 1, 6, "", ("93.184.216.34", 443)), + (2, 1, 6, "", ("10.0.0.4", 443)), + ], + ), self.assertRaises(OutboundHttpBlocked): + validate_outbound_http_url("https://connector.example.test/path", policy=policy) + + def test_bounded_response_rejects_declared_and_streamed_oversize_payloads(self) -> None: + with self.assertRaises(OutboundResponseTooLarge): + bounded_response_bytes(io.BytesIO(b"small"), headers={"Content-Length": "20"}, max_bytes=10) + with self.assertRaises(OutboundResponseTooLarge): + bounded_response_bytes(io.BytesIO(b"eleven-byte"), max_bytes=10) + self.assertEqual(b"ten-bytes!", bounded_response_bytes(io.BytesIO(b"ten-bytes!"), max_bytes=10)) + with patch.dict("os.environ", {"GOVOPLAN_CONNECTOR_MAX_STRUCTURED_RESPONSE_BYTES": "5"}), self.assertRaises( + OutboundResponseTooLarge + ): + bounded_response_bytes(io.BytesIO(b"123456"), max_bytes=10) + + def test_bounded_chunks_rejects_an_oversized_chunk_before_retaining_it(self) -> None: + with self.assertRaises(OutboundResponseTooLarge): + bounded_chunks_bytes((b"one-large-chunk",), max_bytes=8) + + def test_connection_time_resolution_blocks_dns_rebinding_before_socket_open(self) -> None: + policy = outbound_http_policy({"APP_ENV": "production"}) + public = [(2, 1, 6, "", ("93.184.216.34", 443))] + private = [(2, 1, 6, "", ("127.0.0.1", 443))] + with patch( + "govoplan_core.security.outbound_http.socket.getaddrinfo", + side_effect=(public, private), + ) as resolver, patch("govoplan_core.security.outbound_http.socket.socket") as socket_factory: + validate_outbound_http_url("https://connector.example.test/path", policy=policy) + with self.assertRaisesRegex(OutboundHttpBlocked, "non-public network"): + create_outbound_connection("connector.example.test", 443, policy=policy) + + self.assertEqual(2, resolver.call_count) + socket_factory.assert_not_called() + + def test_unpinned_sdk_endpoints_fail_closed_for_dns_hosts(self) -> None: + policy = outbound_http_policy({"APP_ENV": "production"}) + with patch( + "govoplan_core.security.outbound_http.socket.getaddrinfo", + return_value=[(2, 1, 6, "", ("93.184.216.34", 443))], + ), self.assertRaisesRegex(OutboundHttpBlocked, "cannot pin"): + validate_unpinned_sdk_http_url( + "https://objects.example.test", + label="S3 endpoint", + policy=policy, + ) + + def test_core_redirects_strip_credentials_cross_origin_and_reject_https_downgrades(self) -> None: + import urllib.request + + request = urllib.request.Request( + "https://catalog.example.test/releases", + headers={"Authorization": "Bearer secret", "X-Request-ID": "request-1"}, + ) + handler = _PolicyRedirectHandler(label="Catalog URL") + with patch.dict("os.environ", {"APP_ENV": "test"}): + redirected = handler.redirect_request( + request, + None, + 302, + "Found", + {}, + "https://cdn.example.test/releases", + ) + downgrade = handler.redirect_request( + request, + None, + 302, + "Found", + {}, + "http://catalog.example.test/releases", + ) + + self.assertIsNotNone(redirected) + self.assertIsNone(redirected.get_header("Authorization")) + self.assertEqual("request-1", redirected.get_header("X-request-id")) + self.assertIsNone(downgrade) + if __name__ == "__main__": unittest.main()