Harden outbound connector transports

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()),
+            headers=response_headers,
-            body=response.read(),
+            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)

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)

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